Remove Unused BlockingClusterStatePublishResponseHandler (#69112)

This hasn't been used in a long time as far as I can tell.

server/src/main/java/org/elasticsearch/discovery/BlockingClusterStatePublishResponseHandler.java

@@ -1,90 +0,0 @@
-/*
- * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
- * or more contributor license agreements. Licensed under the Elastic License
- * 2.0 and the Server Side Public License, v 1; you may not use this file except
- * in compliance with, at your election, the Elastic License 2.0 or the Server
- * Side Public License, v 1.
- */
-package org.elasticsearch.discovery;
-
-import org.elasticsearch.cluster.node.DiscoveryNode;
-import org.elasticsearch.common.unit.TimeValue;
-import org.elasticsearch.common.util.concurrent.ConcurrentCollections;
-
-import java.util.Collections;
-import java.util.Set;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-
-
-/**
- * Handles responses obtained when publishing a new cluster state from master to all non master nodes.
- * Allows to await a reply from all non master nodes, up to a timeout
- */
-public class BlockingClusterStatePublishResponseHandler {
-
-    private final CountDownLatch latch;
-    private final Set<DiscoveryNode> pendingNodes;
-    private final Set<DiscoveryNode> failedNodes;
-
-    /**
-     * Creates a new BlockingClusterStatePublishResponseHandler
-     * @param publishingToNodes the set of nodes to which the cluster state will be published and should respond
-     */
-    public BlockingClusterStatePublishResponseHandler(Set<DiscoveryNode> publishingToNodes) {
-        this.pendingNodes = ConcurrentCollections.newConcurrentSet();
-        this.pendingNodes.addAll(publishingToNodes);
-        this.latch = new CountDownLatch(pendingNodes.size());
-        this.failedNodes = ConcurrentCollections.newConcurrentSet();
-    }
-
-    /**
-     * Called for each response obtained from non master nodes
-     *
-     * @param node the node that replied to the publish event
-     */
-    public void onResponse(DiscoveryNode node) {
-        boolean found = pendingNodes.remove(node);
-        assert found : "node [" + node + "] already responded or failed";
-        latch.countDown();
-    }
-
-    /**
-     * Called for each failure obtained from non master nodes
-     * @param node the node that replied to the publish event
-     */
-    public void onFailure(DiscoveryNode node) {
-        boolean found = pendingNodes.remove(node);
-        assert found : "node [" + node + "] already responded or failed";
-        boolean added = failedNodes.add(node);
-        assert added : "duplicate failures for " + node;
-        latch.countDown();
-    }
-
-    /**
-     * Allows to wait for all non master nodes to reply to the publish event up to a timeout
-     * @param timeout the timeout
-     * @return true if the timeout expired or not, false otherwise
-     */
-    public boolean awaitAllNodes(TimeValue timeout) throws InterruptedException {
-        boolean success = latch.await(timeout.millis(), TimeUnit.MILLISECONDS);
-        assert success == false || pendingNodes.isEmpty() : "response count reached 0 but still waiting for some nodes";
-        return success;
-    }
-
-    /**
-     * returns a list of nodes which didn't respond yet
-     */
-    public DiscoveryNode[] pendingNodes() {
-        // we use a zero length array, because if we try to pre allocate we may need to remove trailing
-        // nulls if some nodes responded in the meanwhile
-        return pendingNodes.toArray(new DiscoveryNode[0]);
-    }
-
-    /**
-     * returns a set of nodes for which publication has failed.
-     */
-    public Set<DiscoveryNode> getFailedNodes() {
-        return Collections.unmodifiableSet(failedNodes);
-    }
-}

server/src/test/java/org/elasticsearch/discovery/BlockingClusterStatePublishResponseHandlerTests.java

@@ -1,121 +0,0 @@
-/*
- * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
- * or more contributor license agreements. Licensed under the Elastic License
- * 2.0 and the Server Side Public License, v 1; you may not use this file except
- * in compliance with, at your election, the Elastic License 2.0 or the Server
- * Side Public License, v 1.
- */
-package org.elasticsearch.discovery;
-
-import org.apache.logging.log4j.Logger;
-import org.elasticsearch.Version;
-import org.elasticsearch.cluster.node.DiscoveryNode;
-import org.elasticsearch.common.unit.TimeValue;
-import org.elasticsearch.common.util.concurrent.AbstractRunnable;
-import org.elasticsearch.test.ESTestCase;
-
-import java.util.Arrays;
-import java.util.HashSet;
-import java.util.Set;
-import java.util.concurrent.CyclicBarrier;
-
-import static java.util.Collections.emptyMap;
-import static java.util.Collections.emptySet;
-import static org.hamcrest.Matchers.arrayWithSize;
-import static org.hamcrest.Matchers.contains;
-import static org.hamcrest.Matchers.equalTo;
-import static org.hamcrest.Matchers.not;
-
-public class BlockingClusterStatePublishResponseHandlerTests extends ESTestCase {
-
-    private static class PublishResponder extends AbstractRunnable {
-
-        final boolean fail;
-        final DiscoveryNode node;
-        final CyclicBarrier barrier;
-        final Logger logger;
-        final BlockingClusterStatePublishResponseHandler handler;
-
-        PublishResponder(boolean fail, DiscoveryNode node, CyclicBarrier barrier, Logger logger,
-                         BlockingClusterStatePublishResponseHandler handler) {
-            this.fail = fail;
-
-            this.node = node;
-            this.barrier = barrier;
-            this.logger = logger;
-            this.handler = handler;
-        }
-
-        @Override
-        public void onFailure(Exception e) {
-            logger.error("unexpected error", e);
-        }
-
-        @Override
-        protected void doRun() throws Exception {
-            barrier.await();
-            if (fail) {
-                handler.onFailure(node);
-            } else {
-                handler.onResponse(node);
-            }
-        }
-    }
-
-    public void testConcurrentAccess() throws InterruptedException {
-        int nodeCount = scaledRandomIntBetween(10, 20);
-        DiscoveryNode[] allNodes = new DiscoveryNode[nodeCount];
-        for (int i = 0; i < nodeCount; i++) {
-            DiscoveryNode node = new DiscoveryNode("node_" + i, buildNewFakeTransportAddress(), emptyMap(), emptySet(), Version.CURRENT);
-            allNodes[i] = node;
-        }
-
-        BlockingClusterStatePublishResponseHandler handler =
-            new BlockingClusterStatePublishResponseHandler(new HashSet<>(Arrays.asList(allNodes)));
-
-        int firstRound = randomIntBetween(5, nodeCount - 1);
-        Thread[] threads = new Thread[firstRound];
-        CyclicBarrier barrier = new CyclicBarrier(firstRound);
-        Set<DiscoveryNode> expectedFailures = new HashSet<>();
-        Set<DiscoveryNode> completedNodes = new HashSet<>();
-        for (int i = 0; i < threads.length; i++) {
-            final DiscoveryNode node = allNodes[i];
-            completedNodes.add(node);
-            final boolean fail = randomBoolean();
-            if (fail) {
-                expectedFailures.add(node);
-            }
-            threads[i] = new Thread(new PublishResponder(fail, node, barrier, logger, handler));
-            threads[i].start();
-        }
-        // wait on the threads to finish
-        for (Thread t : threads) {
-            t.join();
-        }
-        // verify that the publisher times out
-        assertFalse("expected handler wait to timeout as not all nodes responded", handler.awaitAllNodes(new TimeValue(10)));
-        Set<DiscoveryNode> pendingNodes = new HashSet<>(Arrays.asList(handler.pendingNodes()));
-        assertThat(completedNodes, not(contains(pendingNodes.toArray(new DiscoveryNode[0]))));
-        assertThat(completedNodes.size() + pendingNodes.size(), equalTo(allNodes.length));
-        int secondRound = allNodes.length - firstRound;
-        threads = new Thread[secondRound];
-        barrier = new CyclicBarrier(secondRound);
-
-        for (int i = 0; i < threads.length; i++) {
-            final DiscoveryNode node = allNodes[firstRound + i];
-            final boolean fail = randomBoolean();
-            if (fail) {
-                expectedFailures.add(node);
-            }
-            threads[i] = new Thread(new PublishResponder(fail, node, barrier, logger, handler));
-            threads[i].start();
-        }
-        // wait on the threads to finish
-        for (Thread t : threads) {
-            t.join();
-        }
-        assertTrue("expected handler not to timeout as all nodes responded", handler.awaitAllNodes(new TimeValue(10)));
-        assertThat(handler.pendingNodes(), arrayWithSize(0));
-        assertThat(handler.getFailedNodes(), equalTo(expectedFailures));
-    }
-}