Change runnable to action listener

imRishN · Aug 31, 2022 · 8eb27f0 · 8eb27f0
1 parent 9233757
commit 8eb27f0
Show file tree

Hide file tree

Showing 4 changed files with 27 additions and 24 deletions.
diff --git a/server/src/main/java/org/opensearch/cluster/coordination/Coordinator.java b/server/src/main/java/org/opensearch/cluster/coordination/Coordinator.java
@@ -243,8 +243,7 @@ public Coordinator(
             this.onJoinValidators,
             rerouteService,
             nodeHealthService,
-            this.peerFinder::onDecommission,
-            this.peerFinder::onRecommission
+            peerFinder.nodeCommissionedListener()
         );
         this.publicationHandler = new PublicationTransportHandler(
             transportService,

diff --git a/server/src/main/java/org/opensearch/cluster/coordination/JoinHelper.java b/server/src/main/java/org/opensearch/cluster/coordination/JoinHelper.java
@@ -116,8 +116,7 @@ public class JoinHelper {
     private final NodeHealthService nodeHealthService;
 
     public boolean isDecommissioned;
-    private Runnable onDecommission;
-    private Runnable onRecommission;
+    private final ActionListener<Void> nodeCommissionedListener;
 
     private final Set<Tuple<DiscoveryNode, JoinRequest>> pendingOutgoingJoins = Collections.synchronizedSet(new HashSet<>());
 
@@ -137,13 +136,13 @@ public class JoinHelper {
         Collection<BiConsumer<DiscoveryNode, ClusterState>> joinValidators,
         RerouteService rerouteService,
         NodeHealthService nodeHealthService,
-        Runnable onDecommission,
-        Runnable onRecommission
+        ActionListener<Void> nodeCommissionedListener
     ) {
         this.clusterManagerService = clusterManagerService;
         this.transportService = transportService;
         this.nodeHealthService = nodeHealthService;
         this.joinTimeout = JOIN_TIMEOUT_SETTING.get(settings);
+        this.nodeCommissionedListener = nodeCommissionedListener;
         this.joinTaskExecutorGenerator = () -> new JoinTaskExecutor(settings, allocationService, logger, rerouteService, transportService) {
 
             private final long term = currentTermSupplier.getAsLong();
@@ -353,7 +352,7 @@ public void handleResponse(Empty response) {
                         onCompletion.run();
                         if (isDecommissioned) {
                             isDecommissioned = false;
-                            onRecommission.run();
+                            nodeCommissionedListener.onResponse(null);
                         }
                     }
 
@@ -364,7 +363,7 @@ public void handleException(TransportException exp) {
                             logger.info("local node is decommissioned. Will not be able to join the cluster");
                             if (!isDecommissioned) {
                                 isDecommissioned = true;
-                                onDecommission.run();
+                                nodeCommissionedListener.onFailure(exp);
                             }
                         }
                         logger.info(() -> new ParameterizedMessage("failed to join {} with {}", destination, joinRequest), exp);

diff --git a/server/src/main/java/org/opensearch/discovery/PeerFinder.java b/server/src/main/java/org/opensearch/discovery/PeerFinder.java
@@ -138,6 +138,22 @@ public PeerFinder(
         );
     }
 
+    public ActionListener<Void> nodeCommissionedListener() {
+        return new ActionListener<Void>() {
+            @Override
+            public void onResponse(Void unused) {
+                logger.info("setting findPeersInterval to [{}], due to recommissioning", findPeersInterval);
+                findPeersInterval = DISCOVERY_FIND_PEERS_INTERVAL_SETTING.get(settings);
+            }
+
+            @Override
+            public void onFailure(Exception e) {
+                logger.info("setting findPeersInterval to [{}], due to decommissioning", findPeersInterval);
+                findPeersInterval = DISCOVERY_FIND_PEERS_INTERVAL_DURING_DECOMMISSION_SETTING.get(settings);
+            }
+        };
+    }
+
     public void activate(final DiscoveryNodes lastAcceptedNodes) {
         logger.trace("activating with {}", lastAcceptedNodes);
 
@@ -166,16 +182,6 @@ public void deactivate(DiscoveryNode leader) {
         }
     }
 
-    public void onDecommission() {
-        findPeersInterval = DISCOVERY_FIND_PEERS_INTERVAL_DURING_DECOMMISSION_SETTING.get(settings);
-        logger.info("setting findPeersInterval to [{}], due to decommissioning", findPeersInterval);
-    }
-
-    public void onRecommission() {
-        findPeersInterval = DISCOVERY_FIND_PEERS_INTERVAL_SETTING.get(settings);
-        logger.info("setting findPeersInterval to [{}], due to recommissioning", findPeersInterval);
-    }
-
     // exposed to subclasses for testing
     protected final boolean holdsLock() {
         return Thread.holdsLock(mutex);

diff --git a/server/src/test/java/org/opensearch/cluster/coordination/JoinHelperTests.java b/server/src/test/java/org/opensearch/cluster/coordination/JoinHelperTests.java
@@ -33,6 +33,7 @@
 
 import org.apache.logging.log4j.Level;
 import org.opensearch.Version;
+import org.opensearch.action.ActionListener;
 import org.opensearch.action.ActionListenerResponseHandler;
 import org.opensearch.action.support.PlainActionFuture;
 import org.opensearch.cluster.ClusterName;
@@ -55,6 +56,7 @@
 import java.util.Optional;
 import java.util.concurrent.atomic.AtomicReference;
 
+import static org.mockito.Mockito.mock;
 import static org.opensearch.monitor.StatusInfo.Status.HEALTHY;
 import static org.opensearch.monitor.StatusInfo.Status.UNHEALTHY;
 import static org.opensearch.node.Node.NODE_NAME_SETTING;
@@ -91,8 +93,7 @@ public void testJoinDeduplication() {
             Collections.emptyList(),
             (s, p, r) -> {},
             () -> new StatusInfo(HEALTHY, "info"),
-            ()-> {},
-            () -> {}
+            mock(ActionListener.class)
         );
         transportService.start();
 
@@ -233,8 +234,7 @@ private void assertJoinValidationRejectsMismatchedClusterUUID(String actionName,
             Collections.emptyList(),
             (s, p, r) -> {},
             null,
-            () -> {},
-            () -> {}
+            mock(ActionListener.class)
         ); // registers request handler
         transportService.start();
         transportService.acceptIncomingRequests();
@@ -289,8 +289,7 @@ public void testJoinFailureOnUnhealthyNodes() {
             Collections.emptyList(),
             (s, p, r) -> {},
             () -> nodeHealthServiceStatus.get(),
-            () -> {},
-            () -> {}
+            mock(ActionListener.class)
         );
         transportService.start();