Refactor and make sure to synchronize on only one object

2 files changed, 31 insertions, 48 deletions

diff --git a/config-proxy/src/main/java/com/yahoo/vespa/config/proxy/RpcConfigSourceClient.java b/config-proxy/src/main/java/com/yahoo/vespa/config/proxy/RpcConfigSourceClient.java
index 1f5d49270ed..942d06367b8 100644
--- a/config-proxy/src/main/java/com/yahoo/vespa/config/proxy/RpcConfigSourceClient.java
+++ b/config-proxy/src/main/java/com/yahoo/vespa/config/proxy/RpcConfigSourceClient.java
@@ -2,6 +2,7 @@
 package com.yahoo.vespa.config.proxy;
 
 import com.yahoo.concurrent.DaemonThreadFactory;
+import com.yahoo.config.ConfigurationRuntimeException;
 import com.yahoo.config.subscription.ConfigSourceSet;
 import com.yahoo.config.subscription.impl.JRTConfigRequester;
 import com.yahoo.jrt.Request;
@@ -24,7 +25,7 @@ import java.util.logging.Logger;
  * @author hmusum
  * @since 5.1.9
  */
-public class RpcConfigSourceClient extends ConfigSourceClient {
+class RpcConfigSourceClient extends ConfigSourceClient {
 
     private final static Logger log = Logger.getLogger(RpcConfigSourceClient.class.getName());
     private final Supervisor supervisor = new Supervisor(new Transport());
@@ -140,18 +141,28 @@ public class RpcConfigSourceClient extends ConfigSourceClient {
             delayedResponses.add(new DelayedResponse(request));
         }
         if (needToGetConfig) {
-            synchronized (activeSubscribersLock) {
-                if (activeSubscribers.containsKey(configCacheKey)) {
-                    log.log(LogLevel.DEBUG, "Already a subscriber running for: " + configCacheKey);
-                } else {
-                    log.log(LogLevel.DEBUG, "Could not find good config in cache, creating subscriber for: " + configCacheKey);
-                    Subscriber subscriber = new UpstreamConfigSubscriber(input, clientUpdater, configSourceSet, timingValues, requesterPool, activeSubscribers);
+            subscribeToConfig(input, configCacheKey);
+        }
+        return ret;
+    }
+
+    private void subscribeToConfig(RawConfig input, ConfigCacheKey configCacheKey) {
+        synchronized (activeSubscribersLock) {
+            if (activeSubscribers.containsKey(configCacheKey)) {
+                log.log(LogLevel.DEBUG, "Already a subscriber running for: " + configCacheKey);
+            } else {
+                log.log(LogLevel.DEBUG, "Could not find good config in cache, creating subscriber for: " + configCacheKey);
+                UpstreamConfigSubscriber subscriber = new UpstreamConfigSubscriber(input, clientUpdater, configSourceSet, timingValues, requesterPool);
+                try {
+                    subscriber.subscribe();
                     activeSubscribers.put(configCacheKey, subscriber);
                     exec.execute(subscriber);
+                } catch (ConfigurationRuntimeException e) {
+                    log.log(LogLevel.INFO, "Subscribe for '" + configCacheKey + "' failed, closing subscriber");
+                    subscriber.cancel();
                 }
             }
         }
-        return ret;
     }
 
     @Override
diff --git a/config-proxy/src/main/java/com/yahoo/vespa/config/proxy/UpstreamConfigSubscriber.java b/config-proxy/src/main/java/com/yahoo/vespa/config/proxy/UpstreamConfigSubscriber.java
index 74bbcf1a406..7a8a4436749 100644
--- a/config-proxy/src/main/java/com/yahoo/vespa/config/proxy/UpstreamConfigSubscriber.java
+++ b/config-proxy/src/main/java/com/yahoo/vespa/config/proxy/UpstreamConfigSubscriber.java
@@ -1,7 +1,6 @@
 // Copyright 2016 Yahoo Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 package com.yahoo.vespa.config.proxy;
 
-import com.yahoo.config.ConfigurationRuntimeException;
 import com.yahoo.config.subscription.ConfigSource;
 import com.yahoo.config.subscription.ConfigSourceSet;
 import com.yahoo.config.subscription.impl.GenericConfigHandle;
@@ -9,11 +8,9 @@ import com.yahoo.config.subscription.impl.GenericConfigSubscriber;
 import com.yahoo.config.subscription.impl.JRTConfigRequester;
 import com.yahoo.log.LogLevel;
 import com.yahoo.yolean.Exceptions;
-import com.yahoo.vespa.config.ConfigCacheKey;
 import com.yahoo.vespa.config.RawConfig;
 import com.yahoo.vespa.config.TimingValues;
 
-import java.util.HashMap;
 import java.util.Map;
 import java.util.logging.Logger;
 
@@ -29,58 +26,33 @@ public class UpstreamConfigSubscriber implements Subscriber {
     private final ConfigSource configSourceSet;
     private final TimingValues timingValues;
     private Map<ConfigSourceSet, JRTConfigRequester> requesterPool;
-    private final Map<ConfigCacheKey, Subscriber> activeSubscribers;
     private GenericConfigSubscriber subscriber;
+    private GenericConfigHandle handle;
 
-    public UpstreamConfigSubscriber(RawConfig config,
-                                    ClientUpdater clientUpdater,
-                                    ConfigSource configSourceSet,
-                                    TimingValues timingValues,
-                                    Map<ConfigSourceSet, JRTConfigRequester> requesterPool,
-                                    Map<ConfigCacheKey, Subscriber> activeSubscribers) {
+    UpstreamConfigSubscriber(RawConfig config,
+                             ClientUpdater clientUpdater,
+                             ConfigSource configSourceSet,
+                             TimingValues timingValues,
+                             Map<ConfigSourceSet, JRTConfigRequester> requesterPool) {
         this.config = config;
         this.clientUpdater = clientUpdater;
         this.configSourceSet = configSourceSet;
         this.timingValues = timingValues;
         this.requesterPool = requesterPool;
-        this.activeSubscribers = activeSubscribers;
     }
 
-    UpstreamConfigSubscriber(RawConfig config,
-                             ClientUpdater clientUpdater,
-                             ConfigSource configSourceSet,
-                             TimingValues timingValues,
-                             Map<ConfigSourceSet, JRTConfigRequester> requesterPool) {
-        this(config, clientUpdater, configSourceSet, timingValues, requesterPool, new HashMap<>());
+    void subscribe() {
+        subscriber = new GenericConfigSubscriber(requesterPool);
+        handle = subscriber.subscribe(config.getKey(), config.getDefContent(), configSourceSet, timingValues);
     }
 
     @Override
     public void run() {
-        GenericConfigHandle handle;
-        subscriber = new GenericConfigSubscriber(requesterPool);
-        try {
-            handle = subscriber.subscribe(config.getKey(), config.getDefContent(), configSourceSet, timingValues);
-        } catch (ConfigurationRuntimeException e) {
-            log.log(LogLevel.INFO, "Subscribe for '" + config + "' failed, closing subscriber");
-            final ConfigCacheKey key = new ConfigCacheKey(config.getKey(), config.getDefMd5());
-            synchronized (activeSubscribers) {
-                final Subscriber activeSubscriber = activeSubscribers.get(key);
-                if (activeSubscriber != null) {
-                    activeSubscriber.cancel();
-                    activeSubscribers.remove(key);
-                }
-            }
-            return;
-        }
-
         do {
+            if (! subscriber.nextGeneration()) continue;
+
             try {
-                if (subscriber.nextGeneration()) {
-                    if (log.isLoggable(LogLevel.DEBUG)) {
-                        log.log(LogLevel.DEBUG, "nextGeneration returned for " + config.getKey() + ", subscriber generation=" + subscriber.getGeneration());
-                    }
-                    updateWithNewConfig(handle);
-                }
+                updateWithNewConfig(handle);
             } catch (Exception e) {  // To avoid thread throwing exception and loop never running this again
                 log.log(LogLevel.WARNING, "Got exception: " + Exceptions.toMessageString(e));
             } catch (Throwable e) {