mock admin when needed only

connect/runtime/src/test/java/org/apache/kafka/connect/util/KafkaBasedLogTest.java

@@ -70,6 +70,7 @@
 import static org.mockito.Mockito.any;
 import static org.mockito.Mockito.anyLong;
 import static org.mockito.Mockito.eq;
+import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
@@ -115,12 +116,11 @@ public class KafkaBasedLogTest {
     private final Time time = new MockTime();
     private MockedKafkaBasedLog store;
 
-    @Mock
-    private TopicAdmin admin = null;
     @Mock
     private Consumer<TopicAdmin> initializer;
     @Mock
     private KafkaProducer<String, String> producer;
+    private TopicAdmin admin;
     private final Supplier<TopicAdmin> topicAdminSupplier = () -> admin;
     private MockConsumer<String, String> consumer;
 
@@ -445,6 +445,7 @@ public void testReadEndOffsetsUsingAdmin() {
         Map<TopicPartition, Long> endOffsets = new HashMap<>();
         endOffsets.put(TP0, 0L);
         endOffsets.put(TP1, 0L);
+        admin = mock(TopicAdmin.class);
         when(admin.retryEndOffsets(eq(tps), any(), anyLong())).thenReturn(endOffsets);
         when(admin.endOffsets(eq(tps))).thenReturn(endOffsets);
 
@@ -455,6 +456,7 @@ public void testReadEndOffsetsUsingAdmin() {
     @Test
     public void testReadEndOffsetsUsingAdminThatFailsWithUnsupported() {
         Set<TopicPartition> tps = new HashSet<>(Arrays.asList(TP0, TP1));
+        admin = mock(TopicAdmin.class);
         // Getting end offsets using the admin client should fail with unsupported version
         when(admin.retryEndOffsets(eq(tps), any(), anyLong())).thenThrow(new UnsupportedVersionException("too old"));
 
@@ -474,6 +476,7 @@ public void testReadEndOffsetsUsingAdminThatFailsWithRetriable() {
         Map<TopicPartition, Long> endOffsets = new HashMap<>();
         endOffsets.put(TP0, 0L);
         endOffsets.put(TP1, 0L);
+        admin = mock(TopicAdmin.class);
         // Getting end offsets upon startup should work fine
         when(admin.retryEndOffsets(eq(tps), any(), anyLong())).thenReturn(endOffsets);
         // Getting end offsets using the admin client should fail with leader not available