Set up reconfigurable dispatcher with self-subscription instead

3 files changed, 49 insertions, 7 deletions

diff --git a/config-model/src/main/java/com/yahoo/vespa/model/container/search/ContainerSearch.java b/config-model/src/main/java/com/yahoo/vespa/model/container/search/ContainerSearch.java
index 414d4c817c7..728b4d40bdd 100644
--- a/config-model/src/main/java/com/yahoo/vespa/model/container/search/ContainerSearch.java
+++ b/config-model/src/main/java/com/yahoo/vespa/model/container/search/ContainerSearch.java
@@ -6,6 +6,8 @@ import com.yahoo.container.QrSearchersConfig;
 import com.yahoo.prelude.semantics.SemanticRulesConfig;
 import com.yahoo.search.config.IndexInfoConfig;
 import com.yahoo.search.config.SchemaInfoConfig;
+import com.yahoo.search.dispatch.Dispatcher;
+import com.yahoo.search.dispatch.ReconfigurableDispatcher;
 import com.yahoo.search.pagetemplates.PageTemplatesConfig;
 import com.yahoo.search.query.profile.config.QueryProfilesConfig;
 import com.yahoo.search.ranking.RankProfilesEvaluatorFactory;
@@ -81,16 +83,18 @@ public class ContainerSearch extends ContainerSubsystem<SearchChains>
 
     /** Adds a Dispatcher component to the owning container cluster for each search cluster */
     private void initializeDispatchers(Collection<SearchCluster> searchClusters) {
+        boolean useReconfigurableDispatch = false;
+        Class<? extends Dispatcher> dispatcherClass = useReconfigurableDispatch ? ReconfigurableDispatcher.class : Dispatcher.class;
         for (SearchCluster searchCluster : searchClusters) {
             if (searchCluster instanceof IndexedSearchCluster indexed) {
-                var dispatcher = new DispatcherComponent(indexed);
+                var dispatcher = new DispatcherComponent(indexed, dispatcherClass);
                 owningCluster.addComponent(dispatcher);
             }
             if (globalPhase) {
                 for (var documentDb : searchCluster.getDocumentDbs()) {
-                    if (!schemasWithGlobalPhase.contains(documentDb.getSchemaName())) continue;
+                    if ( ! schemasWithGlobalPhase.contains(documentDb.getSchemaName())) continue;
                     var factory = new RankProfilesEvaluatorComponent(documentDb);
-                    if (! owningCluster.getComponentsMap().containsKey(factory.getComponentId())) {
+                    if ( ! owningCluster.getComponentsMap().containsKey(factory.getComponentId())) {
                         owningCluster.addComponent(factory);
                     }
                 }
diff --git a/config-model/src/main/java/com/yahoo/vespa/model/container/search/DispatcherComponent.java b/config-model/src/main/java/com/yahoo/vespa/model/container/search/DispatcherComponent.java
index f9a3a1f1990..fe2df8101bd 100644
--- a/config-model/src/main/java/com/yahoo/vespa/model/container/search/DispatcherComponent.java
+++ b/config-model/src/main/java/com/yahoo/vespa/model/container/search/DispatcherComponent.java
@@ -3,6 +3,7 @@ package com.yahoo.vespa.model.container.search;
 
 import com.yahoo.config.model.producer.TreeConfigProducer;
 import com.yahoo.osgi.provider.model.ComponentModel;
+import com.yahoo.search.dispatch.Dispatcher;
 import com.yahoo.vespa.config.search.DispatchConfig;
 import com.yahoo.vespa.config.search.DispatchNodesConfig;
 import com.yahoo.vespa.model.container.component.Component;
@@ -22,15 +23,15 @@ public class DispatcherComponent extends Component<TreeConfigProducer<?>, Compon
 
     private final IndexedSearchCluster indexedSearchCluster;
 
-    public DispatcherComponent(IndexedSearchCluster indexedSearchCluster) {
-        super(toComponentModel(indexedSearchCluster.getClusterName()));
+    public DispatcherComponent(IndexedSearchCluster indexedSearchCluster, Class<? extends Dispatcher> clazz) {
+        super(toComponentModel(indexedSearchCluster.getClusterName(), clazz));
         this.indexedSearchCluster = indexedSearchCluster;
     }
 
-    private static ComponentModel toComponentModel(String clusterName) {
+    private static ComponentModel toComponentModel(String clusterName, Class<? extends Dispatcher> clazz) {
         String dispatcherComponentId = "dispatcher." + clusterName; // used by ClusterSearcher
         return new ComponentModel(dispatcherComponentId,
-                                  com.yahoo.search.dispatch.Dispatcher.class.getName(),
+                                  clazz.getName(),
                                   PlatformBundles.SEARCH_AND_DOCPROC_BUNDLE);
     }
 
diff --git a/container-search/src/main/java/com/yahoo/search/dispatch/ReconfigurableDispatcher.java b/container-search/src/main/java/com/yahoo/search/dispatch/ReconfigurableDispatcher.java
new file mode 100644
index 00000000000..625a8bcb6da
--- /dev/null
+++ b/container-search/src/main/java/com/yahoo/search/dispatch/ReconfigurableDispatcher.java
@@ -0,0 +1,37 @@
+package com.yahoo.search.dispatch;
+
+import com.yahoo.component.ComponentId;
+import com.yahoo.config.subscription.ConfigSubscriber;
+import com.yahoo.container.handler.VipStatus;
+import com.yahoo.messagebus.network.rpc.SlobrokConfigSubscriber;
+import com.yahoo.vespa.config.search.DispatchConfig;
+import com.yahoo.vespa.config.search.DispatchNodesConfig;
+import com.yahoo.yolean.UncheckedInterruptedException;
+
+import java.util.Objects;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * @author jonmv
+ */
+public class ReconfigurableDispatcher extends Dispatcher {
+
+    private final ConfigSubscriber subscriber;
+
+    public ReconfigurableDispatcher(ComponentId clusterId, DispatchConfig dispatchConfig, VipStatus vipStatus) {
+        super(clusterId, dispatchConfig, new DispatchNodesConfig.Builder().build(), vipStatus);
+        this.subscriber = new ConfigSubscriber();
+        this.subscriber.subscribe(this::updateWithNewConfig, DispatchNodesConfig.class, clusterId.stringValue());
+    }
+
+    @Override
+    public void deconstruct() {
+        subscriber.close();
+        super.deconstruct();
+    }
+
+}