Refactor a bit mock set up so we do it one place one way.

solr/cross-dc-manager/src/test/org/apache/solr/crossdc/manager/CrossDcMockUtils.java

@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.solr.crossdc.manager;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import org.apache.solr.client.solrj.impl.CloudSolrClient;
+import org.apache.solr.client.solrj.impl.ClusterStateProvider;
+
+/** Shared Mockito fixtures for tests that mock a {@link CloudSolrClient}. */
+public final class CrossDcMockUtils {
+
+  private CrossDcMockUtils() {}
+
+  /**
+   * Mocks a {@link CloudSolrClient} wired with a mocked {@link ClusterStateProvider}.
+   *
+   * <p>{@code SolrMessageProcessor.connectToSolrIfNeeded()} calls {@code
+   * getClusterStateProvider().getLiveNodes()} before processing any request, so any mocked {@link
+   * CloudSolrClient} used with it must supply a state provider or the call spins forever retrying a
+   * {@link NullPointerException}. Fetch the same provider mock back later via {@link
+   * CloudSolrClient#getClusterStateProvider()} if a test needs to verify against it.
+   */
+  public static CloudSolrClient mockConnectedCloudSolrClient() {
+    return configureConnected(mock(CloudSolrClient.class));
+  }
+
+  /**
+   * Wires an existing {@link CloudSolrClient} mock (e.g. one injected via {@code @Mock}) with a
+   * mocked {@link ClusterStateProvider}. See {@link #mockConnectedCloudSolrClient()}.
+   */
+  public static CloudSolrClient configureConnected(CloudSolrClient client) {
+    when(client.getClusterStateProvider()).thenReturn(mock(ClusterStateProvider.class));
+    return client;
+  }
+}

solr/cross-dc-manager/src/test/org/apache/solr/crossdc/manager/consumer/KafkaCrossDcConsumerTest.java

@@ -61,6 +61,7 @@
 import org.apache.solr.crossdc.common.KafkaCrossDcConf;
 import org.apache.solr.crossdc.common.KafkaMirroringSink;
 import org.apache.solr.crossdc.common.MirroredSolrRequest;
+import org.apache.solr.crossdc.manager.CrossDcMockUtils;
 import org.apache.solr.crossdc.manager.messageprocessor.SolrMessageProcessor;
 import org.junit.After;
 import org.junit.Before;
@@ -91,10 +92,9 @@ public static void ensureWorkingMockito() {
   @Before
   public void setUp() {
     kafkaConsumerMock = mock(KafkaConsumer.class);
-    clusterStateProviderMock = mock(ClusterStateProvider.class);
+    solrClientMock = CrossDcMockUtils.mockConnectedCloudSolrClient();
+    clusterStateProviderMock = solrClientMock.getClusterStateProvider();
     doAnswer(inv -> clusterStateProviderIsClosed).when(clusterStateProviderMock).isClosed();
-    solrClientMock = mock(CloudSolrClient.class);
-    doReturn(clusterStateProviderMock).when(solrClientMock).getClusterStateProvider();
     kafkaMirroringSinkMock = mock(KafkaMirroringSink.class);
     messageProcessorMock = mock(SolrMessageProcessor.class);
     conf = testCrossDCConf();

solr/cross-dc-manager/src/test/org/apache/solr/crossdc/manager/messageprocessor/SolrMessageProcessorTest.java

@@ -37,6 +37,7 @@
 import org.apache.solr.crossdc.common.IQueueHandler;
 import org.apache.solr.crossdc.common.MirroredSolrRequest;
 import org.apache.solr.crossdc.common.ResubmitBackoffPolicy;
+import org.apache.solr.crossdc.manager.CrossDcMockUtils;
 import org.apache.solr.crossdc.manager.consumer.OtelMetrics;
 import org.junit.Before;
 import org.junit.BeforeClass;
@@ -56,10 +57,8 @@ public static void ensureWorkingMockito() {
 
   @Before
   public void setUp() {
-    client = mock(CloudSolrClient.class);
-    // handleItem() probes the cluster through the state provider, so the mock must supply one
-    clusterStateProvider = mock(ClusterStateProvider.class);
-    when(client.getClusterStateProvider()).thenReturn(clusterStateProvider);
+    client = CrossDcMockUtils.mockConnectedCloudSolrClient();
+    clusterStateProvider = client.getClusterStateProvider();
     resubmitBackoffPolicy = mock(ResubmitBackoffPolicy.class);
     solrMessageProcessor =
         new SolrMessageProcessor(mock(OtelMetrics.class), () -> client, resubmitBackoffPolicy);

solr/cross-dc-manager/src/test/org/apache/solr/crossdc/manager/messageprocessor/TestMessageProcessor.java

@@ -38,20 +38,19 @@
 import org.apache.solr.crossdc.common.IQueueHandler;
 import org.apache.solr.crossdc.common.MirroredSolrRequest;
 import org.apache.solr.crossdc.common.ResubmitBackoffPolicy;
+import org.apache.solr.crossdc.manager.CrossDcMockUtils;
 import org.apache.solr.crossdc.manager.consumer.ConsumerMetrics;
 import org.apache.solr.crossdc.manager.consumer.OtelMetrics;
 import org.junit.Before;
 import org.junit.BeforeClass;
 import org.junit.Ignore;
 import org.junit.Test;
-import org.mockito.Mock;
 import org.mockito.Mockito;
-import org.mockito.MockitoAnnotations;
 
 public class TestMessageProcessor {
   static final String VERSION_FIELD = "_version_";
 
-  @Mock private CloudSolrClient solrClient;
+  private CloudSolrClient solrClient;
   private SolrMessageProcessor processor;
 
   private final ResubmitBackoffPolicy backoffPolicy =
@@ -70,7 +69,7 @@ public static void ensureWorkingMockito() {
 
   @Before
   public void setUp() {
-    MockitoAnnotations.initMocks(this);
+    solrClient = CrossDcMockUtils.mockConnectedCloudSolrClient();
 
     ConsumerMetrics metrics = Mockito.mock(OtelMetrics.class);
     processor = Mockito.spy(new SolrMessageProcessor(metrics, () -> solrClient, backoffPolicy));