Add and expose a StripedExecutor from the TenantRepository

4 files changed, 38 insertions, 5 deletions

diff --git a/configserver/src/main/java/com/yahoo/vespa/config/server/GlobalComponentRegistry.java b/configserver/src/main/java/com/yahoo/vespa/config/server/GlobalComponentRegistry.java
index 2f53973f5af..28387a3c3e6 100644
--- a/configserver/src/main/java/com/yahoo/vespa/config/server/GlobalComponentRegistry.java
+++ b/configserver/src/main/java/com/yahoo/vespa/config/server/GlobalComponentRegistry.java
@@ -2,8 +2,10 @@
 package com.yahoo.vespa.config.server;
 
 import com.yahoo.cloud.config.ConfigserverConfig;
+import com.yahoo.concurrent.StripedExecutor;
 import com.yahoo.config.model.api.ConfigDefinitionRepo;
 import com.yahoo.config.provision.Provisioner;
+import com.yahoo.config.provision.TenantName;
 import com.yahoo.config.provision.Zone;
 import com.yahoo.vespa.config.server.application.PermanentApplicationPackage;
 import com.yahoo.vespa.config.server.host.HostRegistries;
@@ -40,6 +42,7 @@ public interface GlobalComponentRegistry {
     Zone getZone();
     Clock getClock();
     ConfigServerDB getConfigServerDB();
+    StripedExecutor<TenantName> getZkWatcherExecutor();
     FlagSource getFlagSource();
 
 }
diff --git a/configserver/src/main/java/com/yahoo/vespa/config/server/InjectedGlobalComponentRegistry.java b/configserver/src/main/java/com/yahoo/vespa/config/server/InjectedGlobalComponentRegistry.java
index f8e53350e5c..5a1d9aa57c9 100644
--- a/configserver/src/main/java/com/yahoo/vespa/config/server/InjectedGlobalComponentRegistry.java
+++ b/configserver/src/main/java/com/yahoo/vespa/config/server/InjectedGlobalComponentRegistry.java
@@ -3,8 +3,10 @@ package com.yahoo.vespa.config.server;
 
 import com.google.inject.Inject;
 import com.yahoo.cloud.config.ConfigserverConfig;
+import com.yahoo.concurrent.StripedExecutor;
 import com.yahoo.config.model.api.ConfigDefinitionRepo;
 import com.yahoo.config.provision.Provisioner;
+import com.yahoo.config.provision.TenantName;
 import com.yahoo.config.provision.Zone;
 import com.yahoo.vespa.config.server.application.PermanentApplicationPackage;
 import com.yahoo.vespa.config.server.host.HostRegistries;
@@ -42,6 +44,7 @@ public class InjectedGlobalComponentRegistry implements GlobalComponentRegistry
     private final Zone zone;
     private final ConfigServerDB configServerDB;
     private final FlagSource flagSource;
+    private final StripedExecutor<TenantName> zkWatcherExecutor;
 
     @SuppressWarnings("WeakerAccess")
     @Inject
@@ -74,6 +77,7 @@ public class InjectedGlobalComponentRegistry implements GlobalComponentRegistry
         this.zone = zone;
         this.configServerDB = configServerDB;
         this.flagSource = flagSource;
+        this.zkWatcherExecutor = new StripedExecutor<>();
     }
 
     @Override
@@ -116,5 +120,10 @@ public class InjectedGlobalComponentRegistry implements GlobalComponentRegistry
     public ConfigServerDB getConfigServerDB() { return configServerDB; }
 
     @Override
+    public StripedExecutor<TenantName> getZkWatcherExecutor() {
+        return zkWatcherExecutor;
+    }
+
+    @Override
     public FlagSource getFlagSource() { return flagSource; }
 }
diff --git a/configserver/src/main/java/com/yahoo/vespa/config/server/tenant/TenantRepository.java b/configserver/src/main/java/com/yahoo/vespa/config/server/tenant/TenantRepository.java
index bedf481fdc4..ad6b54c7289 100644
--- a/configserver/src/main/java/com/yahoo/vespa/config/server/tenant/TenantRepository.java
+++ b/configserver/src/main/java/com/yahoo/vespa/config/server/tenant/TenantRepository.java
@@ -4,6 +4,7 @@ package com.yahoo.vespa.config.server.tenant;
 import com.google.common.collect.ImmutableSet;
 import com.google.inject.Inject;
 import com.yahoo.cloud.config.ConfigserverConfig;
+import com.yahoo.concurrent.StripedExecutor;
 import com.yahoo.concurrent.ThreadFactoryFactory;
 import com.yahoo.config.provision.ApplicationId;
 import com.yahoo.config.provision.TenantName;
@@ -73,7 +74,8 @@ public class TenantRepository {
     private final Curator curator;
 
     private final MetricUpdater metricUpdater;
-    private final ExecutorService pathChildrenExecutor = Executors.newFixedThreadPool(1, ThreadFactoryFactory.getThreadFactory(TenantRepository.class.getName()));
+    private final ExecutorService zkCacheExecutor = Executors.newFixedThreadPool(1, ThreadFactoryFactory.getThreadFactory(TenantRepository.class.getName()));
+    private final StripedExecutor<TenantName> zkWatcherExecutor;
     private final ExecutorService bootstrapExecutor;
     private final ScheduledExecutorService checkForRemovedApplicationsService = new ScheduledThreadPoolExecutor(1);
     private final Optional<Curator.DirectoryCache> directoryCache;
@@ -104,6 +106,7 @@ public class TenantRepository {
         this.curator = globalComponentRegistry.getCurator();
         metricUpdater = globalComponentRegistry.getMetrics().getOrCreateMetricUpdater(Collections.emptyMap());
         this.tenantListeners.add(globalComponentRegistry.getTenantListener());
+        this.zkWatcherExecutor = globalComponentRegistry.getZkWatcherExecutor();
         curator.framework().getConnectionStateListenable().addListener(this::stateChanged);
 
         curator.create(tenantsPath);
@@ -112,7 +115,7 @@ public class TenantRepository {
         curator.create(vespaPath);
 
         if (useZooKeeperWatchForTenantChanges) {
-            this.directoryCache = Optional.of(curator.createDirectoryCache(tenantsPath.getAbsolute(), false, false, pathChildrenExecutor));
+            this.directoryCache = Optional.of(curator.createDirectoryCache(tenantsPath.getAbsolute(), false, false, zkCacheExecutor));
             this.directoryCache.get().start();
             this.directoryCache.get().addListener(this::childEvent);
         } else {
@@ -158,7 +161,7 @@ public class TenantRepository {
 
     private void checkForRemovedTenants(Set<TenantName> newTenants) {
         for (TenantName tenantName : ImmutableSet.copyOf(tenants.keySet())) {
-            if (!newTenants.contains(tenantName)) {
+            if ( ! newTenants.contains(tenantName)) {
                 closeTenant(tenantName);
             }
         }
@@ -357,12 +360,18 @@ public class TenantRepository {
         }
     }
 
+    /** Use this executor for ZK cache listeners, and have them delegate their work to the {@link #zkWatcherExecutor()}. */
+    public ExecutorService zkCacheExecutor() { return zkCacheExecutor; }
+
+    /** Use this executor to run watcher reactions serially for each tenant. Used by {@link #zkCacheExecutor()}. */
+    public StripedExecutor<TenantName> zkWatcherExecutor() { return zkWatcherExecutor; }
+
     public void close() {
         directoryCache.ifPresent(Curator.DirectoryCache::close);
         try {
-            pathChildrenExecutor.shutdown();
+            zkCacheExecutor.shutdown();
             checkForRemovedApplicationsService.shutdown();
-            pathChildrenExecutor.awaitTermination(50, TimeUnit.SECONDS);
+            zkCacheExecutor.awaitTermination(50, TimeUnit.SECONDS);
             checkForRemovedApplicationsService.awaitTermination(50, TimeUnit.SECONDS);
         }
         catch (InterruptedException e) {
diff --git a/configserver/src/test/java/com/yahoo/vespa/config/server/TestComponentRegistry.java b/configserver/src/test/java/com/yahoo/vespa/config/server/TestComponentRegistry.java
index e369d5bd3bb..0d37cc2f7c9 100644
--- a/configserver/src/test/java/com/yahoo/vespa/config/server/TestComponentRegistry.java
+++ b/configserver/src/test/java/com/yahoo/vespa/config/server/TestComponentRegistry.java
@@ -3,9 +3,12 @@ package com.yahoo.vespa.config.server;
 
 import com.google.common.io.Files;
 import com.yahoo.cloud.config.ConfigserverConfig;
+import com.yahoo.concurrent.InThreadExecutorService;
+import com.yahoo.concurrent.StripedExecutor;
 import com.yahoo.config.model.NullConfigModelRegistry;
 import com.yahoo.config.model.api.ConfigDefinitionRepo;
 import com.yahoo.config.provision.Provisioner;
+import com.yahoo.config.provision.TenantName;
 import com.yahoo.config.provision.Zone;
 import com.yahoo.vespa.config.server.application.PermanentApplicationPackage;
 import com.yahoo.vespa.config.server.host.HostRegistries;
@@ -28,6 +31,7 @@ import com.yahoo.vespa.model.VespaModelFactory;
 import java.time.Clock;
 import java.util.Collections;
 import java.util.Optional;
+import java.util.concurrent.Executors;
 
 
 /**
@@ -51,6 +55,7 @@ public class TestComponentRegistry implements GlobalComponentRegistry {
     private final Zone zone;
     private final Clock clock;
     private final ConfigServerDB configServerDB;
+    private final StripedExecutor<TenantName> zkWatcherExecutor;
 
     private TestComponentRegistry(Curator curator, ConfigCurator configCurator, Metrics metrics,
                                   ModelFactoryRegistry modelFactoryRegistry,
@@ -81,6 +86,7 @@ public class TestComponentRegistry implements GlobalComponentRegistry {
         this.zone = zone;
         this.clock = clock;
         this.configServerDB = new ConfigServerDB(configserverConfig);
+        this.zkWatcherExecutor = new StripedExecutor<>(new InThreadExecutorService());
     }
 
     public static class Builder {
@@ -194,6 +200,12 @@ public class TestComponentRegistry implements GlobalComponentRegistry {
     public Clock getClock() { return clock;}
     @Override
     public ConfigServerDB getConfigServerDB() { return configServerDB;}
+
+    @Override
+    public StripedExecutor<TenantName> getZkWatcherExecutor() {
+        return zkWatcherExecutor;
+    }
+
     @Override
     public FlagSource getFlagSource() { return new InMemoryFlagSource(); }