Replace mocks with test implementation in RepositoriesServiceTests (e…

…lastic#108589)

I observed that `testRegisterRepositorySuccessAfterCreationFailed` test
never invokes assertion blocks, because listener is not invoked.

There are 2 problems:

1. Test setup used mocks. Mocks interrupt listener chain propagation, so registerRepository never returned Response or Failure.
2. We silently ignore assertions in listener because it is not invoked. Test pass successfully.

PutRepositories method relies on cluster state update. I replace mocked
ClusterService and ThreadPool with test implementation of these. Also
add blocking call on listener to ensure we get result.

Address
[comment](elastic#108531 (review))
to break down larger PR into smaller pieces in elastic#108531

server/src/test/java/org/elasticsearch/repositories/RepositoriesServiceTests.java

@@ -10,6 +10,7 @@
 
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.admin.cluster.repositories.put.PutRepositoryRequest;
+import org.elasticsearch.action.support.SubscribableListener;
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
 import org.elasticsearch.cluster.ClusterChangedEvent;
 import org.elasticsearch.cluster.ClusterName;
@@ -20,7 +21,6 @@
 import org.elasticsearch.cluster.metadata.RepositoryMetadata;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.node.DiscoveryNodeUtils;
-import org.elasticsearch.cluster.service.ClusterApplierService;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.UUIDs;
 import org.elasticsearch.common.blobstore.BlobPath;
@@ -29,8 +29,8 @@
 import org.elasticsearch.common.component.LifecycleListener;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.MockBigArrays;
-import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.core.CheckedConsumer;
+import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.index.IndexVersion;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.index.snapshots.IndexShardSnapshotStatus;
@@ -41,11 +41,15 @@
 import org.elasticsearch.snapshots.SnapshotDeleteListener;
 import org.elasticsearch.snapshots.SnapshotId;
 import org.elasticsearch.snapshots.SnapshotInfo;
+import org.elasticsearch.test.ClusterServiceUtils;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.threadpool.TestThreadPool;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.Transport;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xcontent.NamedXContentRegistry;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
 
 import java.util.Arrays;
 import java.util.Collection;
@@ -58,21 +62,31 @@
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.isA;
 import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
 
 public class RepositoriesServiceTests extends ESTestCase {
 
+    private static ThreadPool threadPool;
+
+    private ClusterService clusterService;
     private RepositoriesService repositoriesService;
 
+    @BeforeClass
+    public static void createThreadPool() {
+        threadPool = new TestThreadPool(RepositoriesService.class.getName());
+    }
+
+    @AfterClass
+    public static void terminateThreadPool() {
+        if (threadPool != null) {
+            threadPool.shutdownNow();
+            threadPool = null;
+        }
+    }
+
     @Override
     public void setUp() throws Exception {
         super.setUp();
-        ThreadContext threadContext = new ThreadContext(Settings.EMPTY);
-        ThreadPool threadPool = mock(ThreadPool.class);
-        when(threadPool.getThreadContext()).thenReturn(threadContext);
-        when(threadPool.info(ThreadPool.Names.SNAPSHOT)).thenReturn(
-            new ThreadPool.Info(ThreadPool.Names.SNAPSHOT, ThreadPool.ThreadPoolType.FIXED, randomIntBetween(1, 10))
-        );
+
         final TransportService transportService = new TransportService(
             Settings.EMPTY,
             mock(Transport.class),
@@ -82,10 +96,18 @@ public void setUp() throws Exception {
             null,
             Collections.emptySet()
         );
-        final ClusterApplierService clusterApplierService = mock(ClusterApplierService.class);
-        when(clusterApplierService.threadPool()).thenReturn(threadPool);
-        final ClusterService clusterService = mock(ClusterService.class);
-        when(clusterService.getClusterApplierService()).thenReturn(clusterApplierService);
+
+        clusterService = ClusterServiceUtils.createClusterService(threadPool);
+
+        // cluster utils publisher does not call AckListener, making some method calls hang indefinitely
+        // in this test we have a single master node, and it acknowledges cluster state immediately
+        final var publisher = ClusterServiceUtils.createClusterStatePublisher(clusterService.getClusterApplierService());
+        clusterService.getMasterService().setClusterStatePublisher((evt, pub, ack) -> {
+            publisher.publish(evt, pub, ack);
+            ack.onCommit(TimeValue.ZERO);
+            ack.onNodeAck(clusterService.localNode(), null);
+        });
+
         Map<String, Repository.Factory> typesRegistry = Map.of(
             TestRepository.TYPE,
             TestRepository::new,
@@ -98,16 +120,25 @@ public void setUp() throws Exception {
         );
         repositoriesService = new RepositoriesService(
             Settings.EMPTY,
-            mock(ClusterService.class),
+            clusterService,
             transportService,
             typesRegistry,
             typesRegistry,
             threadPool,
             List.of()
         );
+
+        clusterService.start();
         repositoriesService.start();
     }
 
+    @Override
+    public void tearDown() throws Exception {
+        super.tearDown();
+        clusterService.stop();
+        repositoriesService.stop();
+    }
+
     public void testRegisterInternalRepository() {
         String repoName = "name";
         expectThrows(RepositoryMissingException.class, () -> repositoriesService.repository(repoName));
@@ -283,18 +314,11 @@ public void testRegisterRepositorySuccessAfterCreationFailed() {
         // 2. repository creation successfully when current node become master node and repository is put again
         var request = new PutRepositoryRequest().name(repoName).type(TestRepository.TYPE);
 
-        repositoriesService.registerRepository(request, new ActionListener<>() {
-            @Override
-            public void onResponse(AcknowledgedResponse acknowledgedResponse) {
-                assertTrue(acknowledgedResponse.isAcknowledged());
-                assertThat(repositoriesService.repository(repoName), isA(TestRepository.class));
-            }
-
-            @Override
-            public void onFailure(Exception e) {
-                assert false : e;
-            }
-        });
+        var resultListener = new SubscribableListener<AcknowledgedResponse>();
+        repositoriesService.registerRepository(request, resultListener);
+        var response = safeAwait(resultListener);
+        assertTrue(response.isAcknowledged());
+        assertThat(repositoriesService.repository(repoName), isA(TestRepository.class));
     }
 
     private ClusterState createClusterStateWithRepo(String repoName, String repoType) {
@@ -320,11 +344,10 @@ private void assertThrowsOnRegister(String repoName) {
     private static class TestRepository implements Repository {
 
         private static final String TYPE = "internal";
+        private final RepositoryMetadata metadata;
         private boolean isClosed;
         private boolean isStarted;
 
-        private final RepositoryMetadata metadata;
-
         private TestRepository(RepositoryMetadata metadata) {
             this.metadata = metadata;
         }
@@ -357,7 +380,7 @@ public IndexMetadata getSnapshotIndexMetaData(RepositoryData repositoryData, Sna
 
         @Override
         public void getRepositoryData(Executor responseExecutor, ActionListener<RepositoryData> listener) {
-            listener.onResponse(null);
+            listener.onResponse(RepositoryData.EMPTY);
         }
 
         @Override