Revert "Use a single-threaded executor that delegates to another for ZK sessi…"

4 files changed, 11 insertions, 51 deletions

diff --git a/configserver/src/main/java/com/yahoo/vespa/config/server/session/SessionRepository.java b/configserver/src/main/java/com/yahoo/vespa/config/server/session/SessionRepository.java
index 91ef14e3394..181ed880fd7 100644
--- a/configserver/src/main/java/com/yahoo/vespa/config/server/session/SessionRepository.java
+++ b/configserver/src/main/java/com/yahoo/vespa/config/server/session/SessionRepository.java
@@ -5,6 +5,7 @@ import com.google.common.collect.HashMultiset;
 import com.google.common.collect.Multiset;
 import com.yahoo.cloud.config.ConfigserverConfig;
 import com.yahoo.concurrent.DaemonThreadFactory;
+import com.yahoo.concurrent.StripedExecutor;
 import com.yahoo.config.FileReference;
 import com.yahoo.config.application.api.ApplicationPackage;
 import com.yahoo.config.application.api.DeployLogger;
@@ -127,7 +128,7 @@ public class SessionRepository {
                              SessionPreparer sessionPreparer,
                              ConfigCurator configCurator,
                              Metrics metrics,
-                             ExecutorService zkWatcherExecutor,
+                             StripedExecutor<TenantName> zkWatcherExecutor,
                              PermanentApplicationPackage permanentApplicationPackage,
                              FlagSource flagSource,
                              ExecutorService zkCacheExecutor,
@@ -147,7 +148,7 @@ public class SessionRepository {
         this.clock = clock;
         this.curator = configCurator.curator();
         this.sessionLifetime = Duration.ofSeconds(configserverConfig.sessionLifetime());
-        this.zkWatcherExecutor = zkWatcherExecutor;
+        this.zkWatcherExecutor = command -> zkWatcherExecutor.execute(tenantName, command);
         this.permanentApplicationPackage = permanentApplicationPackage;
         this.flagSource = flagSource;
         this.tenantFileSystemDirs = new TenantFileSystemDirs(configServerDB, tenantName);
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 e62812af494..52d1484897c 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
@@ -58,10 +58,8 @@ import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
-import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
-import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.logging.Level;
 import java.util.logging.Logger;
@@ -105,7 +103,7 @@ public class TenantRepository {
     private final Metrics metrics;
     private final MetricUpdater metricUpdater;
     private final ExecutorService zkCacheExecutor;
-    private final ExecutorService zkSessionWatcherExecutor;
+    private final StripedExecutor<TenantName> zkSessionWatcherExecutor;
     private final StripedExecutor<TenantName> zkApplicationWatcherExecutor;
     private final FileDistributionFactory fileDistributionFactory;
     private final FlagSource flagSource;
@@ -144,7 +142,7 @@ public class TenantRepository {
              configCurator,
              metrics,
              new StripedExecutor<>(),
-             new ZkWatcherExecutorService(),
+             new StripedExecutor<>(),
              new FileDistributionFactory(configserverConfig),
              flagSource,
              Executors.newFixedThreadPool(1, ThreadFactoryFactory.getThreadFactory(TenantRepository.class.getName())),
@@ -164,7 +162,7 @@ public class TenantRepository {
                             ConfigCurator configCurator,
                             Metrics metrics,
                             StripedExecutor<TenantName> zkApplicationWatcherExecutor ,
-                            ExecutorService zkSessionWatcherExecutor,
+                            StripedExecutor<TenantName> zkSessionWatcherExecutor,
                             FileDistributionFactory fileDistributionFactory,
                             FlagSource flagSource,
                             ExecutorService zkCacheExecutor,
@@ -544,10 +542,9 @@ public class TenantRepository {
             zkCacheExecutor.shutdown();
             checkForRemovedApplicationsService.shutdown();
             zkApplicationWatcherExecutor.shutdownAndWait();
-            zkSessionWatcherExecutor.shutdown();
+            zkSessionWatcherExecutor.shutdownAndWait();
             zkCacheExecutor.awaitTermination(50, TimeUnit.SECONDS);
             checkForRemovedApplicationsService.awaitTermination(50, TimeUnit.SECONDS);
-            zkSessionWatcherExecutor.awaitTermination(50, TimeUnit.SECONDS);
         }
         catch (InterruptedException e) {
             log.log(Level.WARNING, "Interrupted while shutting down.", e);
@@ -615,42 +612,4 @@ public class TenantRepository {
 
     public Curator getCurator() { return curator; }
 
-    /**
-     * Single-threaded executor (to make sure that getting data for a ZooKeeper watcher event is done serially) that
-     * delegates to another executor (that uses a cached thread pool)
-     */
-    private static class ZkWatcherExecutorService extends ThreadPoolExecutor {
-
-        private final ExecutorService executorService =
-                Executors.newCachedThreadPool(new DaemonThreadFactory("zk-session-watcher-"));
-
-        public ZkWatcherExecutorService() {
-            super(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>());
-        }
-
-        @Override
-        public void execute(Runnable command) {
-            super.execute(() -> executorService.execute(command));
-        }
-
-        @Override
-        public void shutdown() {
-            super.shutdown();
-            executorService.shutdown();
-        }
-
-        @Override
-        public List<Runnable> shutdownNow() {
-            super.shutdownNow();
-            return executorService.shutdownNow();
-        }
-
-        @Override
-        public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
-            if ( ! super.awaitTermination(timeout, unit)) return false;
-            return executorService.awaitTermination(timeout, unit);
-        }
-
-    }
-
 }
diff --git a/configserver/src/test/java/com/yahoo/vespa/config/server/tenant/TenantRepositoryTest.java b/configserver/src/test/java/com/yahoo/vespa/config/server/tenant/TenantRepositoryTest.java
index 36ce3eaac32..6f7e0541cc7 100644
--- a/configserver/src/test/java/com/yahoo/vespa/config/server/tenant/TenantRepositoryTest.java
+++ b/configserver/src/test/java/com/yahoo/vespa/config/server/tenant/TenantRepositoryTest.java
@@ -172,8 +172,8 @@ public class TenantRepositoryTest {
         try {
             tenantRepository.addTenant(newTenant);
             // Poll for the watcher to pick up the tenant from zk, and add it
-            int tries = 0;
-            while (true) {
+            int tries=0;
+            while(true) {
                 if (tries > 5000) fail("Didn't react on watch");
                 if (tenantRepository.getAllTenantNames().containsAll(expectedTenants)) {
                     break;
@@ -212,7 +212,7 @@ public class TenantRepositoryTest {
                   ConfigCurator.create(new MockCurator()),
                   Metrics.createTestMetrics(),
                   new StripedExecutor<>(new InThreadExecutorService()),
-                  new InThreadExecutorService(),
+                  new StripedExecutor<>(new InThreadExecutorService()),
                   new FileDistributionFactory(new ConfigserverConfig.Builder().build()),
                   new InMemoryFlagSource(),
                   new InThreadExecutorService(),
diff --git a/configserver/src/test/java/com/yahoo/vespa/config/server/tenant/TestTenantRepository.java b/configserver/src/test/java/com/yahoo/vespa/config/server/tenant/TestTenantRepository.java
index 40f046b8b05..687d58fd23b 100644
--- a/configserver/src/test/java/com/yahoo/vespa/config/server/tenant/TestTenantRepository.java
+++ b/configserver/src/test/java/com/yahoo/vespa/config/server/tenant/TestTenantRepository.java
@@ -50,7 +50,7 @@ public class TestTenantRepository extends TenantRepository {
               ConfigCurator.create(curator),
               metrics,
               new StripedExecutor<>(new InThreadExecutorService()),
-              new InThreadExecutorService(),
+              new StripedExecutor<>(new InThreadExecutorService()),
               fileDistributionFactory,
               flagSource,
               new InThreadExecutorService(),