[GitHub] [kafka] yashmayya commented on a diff in pull request #14102: KAFKA-13187: Replace EasyMock / PowerMock with Mockito in DistributedHerderTest

["yashmayya (via GitHub)" <gi...@apache.org>]

yashmayya commented on code in PR #14102:
URL: https://github.com/apache/kafka/pull/14102#discussion_r1287941830


##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java:
##########
@@ -687,101 +617,72 @@ public void revokeAndReassign(boolean incompleteRebalance) throws TimeoutExcepti
             );
             expectConfigRefreshAndSnapshot(secondSnapshot);
         }
-        member.requestRejoin();
-        PowerMock.expectLastCall();
+
+        doNothing().when(member).requestRejoin();
+
+        herder.tick();
 
         // re-assign the connector back to the same worker to ensure state was cleaned up
         expectRebalance(configOffset, Arrays.asList(CONN1), Arrays.asList());
-        Capture<Callback<TargetState>> onSecondStart = newCapture();
-        worker.startConnector(EasyMock.eq(CONN1), EasyMock.anyObject(),
-            EasyMock.anyObject(),
-            EasyMock.eq(herder), EasyMock.eq(TargetState.STARTED), capture(onSecondStart));
-        PowerMock.expectLastCall().andAnswer(() -> {
-            onSecondStart.getValue().onCompletion(null, TargetState.STARTED);
-            return true;
-        });
-        member.wakeup();
-        PowerMock.expectLastCall();
-        expectExecuteTaskReconfiguration(true, conn1SinkConfig, () -> TASK_CONFIGS);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
-
-        PowerMock.replayAll();
 
         herder.tick();
-        if (incompleteRebalance) {
-            herder.tick();
-        }
-        herder.tick();
-        herder.tick();
 
-        PowerMock.verifyAll();
+        verify(worker, times(2)).startConnector(eq(CONN1), any(), any(), eq(herder), eq(TargetState.STARTED), any());
+        verify(worker, times(2)).connectorTaskConfigs(eq(CONN1), eq(conn1SinkConfig));
+        verify(worker).startSourceTask(eq(TASK1), any(), any(), any(), eq(herder), eq(TargetState.STARTED));
+

Review Comment:
   Wow, good catch, and thanks for the additional context!



##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java:
##########
@@ -844,73 +737,57 @@ public void testCreateConnectorConfigBackingStoreError() {
         time.sleep(1000L);
         assertStatistics(3, 1, 100, 1000L);
 
-        PowerMock.verifyAll();
+        // Verify that the exception thrown during the config backing store write is propagated via the callback
+        verify(putConnectorCallback).onCompletion(any(ConnectException.class), isNull());
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore, putConnectorCallback);
     }
 
     @Test
     public void testCreateConnectorFailedValidation() throws Exception {
-        EasyMock.expect(member.memberId()).andStubReturn("leader");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+        when(member.memberId()).thenReturn("leader");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
         expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true);
         expectConfigRefreshAndSnapshot(SNAPSHOT);
 
+        when(statusBackingStore.connectors()).thenReturn(Collections.emptySet());
+        doNothing().when(member).poll(anyLong());
+
+        // Initial rebalance where this member becomes the leader
+        herder.tick();

Review Comment:
   Hm sorry, I don't really remember the context as it's actually been a few months since I worked on this test. Playing around it with it a bit though, it looks like we don't really need the extra tick and I've modified the test accordingly. My guess is that I added the new tick to ensure that the number of rebalances is as expected. The number depends on the ordering of tick calls and `WorkerGroupMember::ensureActive` stubs, but the result can also be achieved by some simpler re-ordering instead (which is what I've done now).



##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java:
##########
@@ -1466,560 +1032,421 @@ public void testRestartConnectorAndTasksUnknownConnector() throws Exception {
         assertTrue(ee.getCause() instanceof NotFoundException);
         assertTrue(ee.getMessage().contains("Unknown connector:"));
 
-        PowerMock.verifyAll();
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testRestartConnectorAndTasksNotLeader() throws Exception {
-        RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
-
         // get the initial assignment
-        EasyMock.expect(member.memberId()).andStubReturn("member");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+        when(member.memberId()).thenReturn("member");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
         expectRebalance(1, Collections.emptyList(), Collections.emptyList());
         expectConfigRefreshAndSnapshot(SNAPSHOT);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
+        doNothing().when(member).poll(anyLong());
 
-        // now handle the connector restart
-        member.wakeup();
-        PowerMock.expectLastCall();
-        member.ensureActive();
-        PowerMock.expectLastCall();
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
+        herder.tick();
 
-        PowerMock.replayAll();
+        // now handle the connector restart
+        doNothing().when(member).ensureActive();
 
         herder.tick();
         FutureCallback<ConnectorStateInfo> callback = new FutureCallback<>();
+        RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
         herder.restartConnectorAndTasks(restartRequest, callback);
         herder.tick();
         ExecutionException ee = assertThrows(ExecutionException.class, () -> callback.get(1000L, TimeUnit.MILLISECONDS));
         assertTrue(ee.getCause() instanceof NotLeaderException);
 
-        PowerMock.verifyAll();
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testRestartConnectorAndTasksUnknownStatus() throws Exception {
-        RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
-        EasyMock.expect(herder.buildRestartPlan(restartRequest)).andReturn(Optional.empty()).anyTimes();
-
-        configBackingStore.putRestartRequest(restartRequest);
-        PowerMock.expectLastCall();
-
         // get the initial assignment
-        EasyMock.expect(member.memberId()).andStubReturn("leader");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+        when(member.memberId()).thenReturn("leader");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
         expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true);
         expectConfigRefreshAndSnapshot(SNAPSHOT);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
-
-        // now handle the connector restart
-        member.wakeup();
-        PowerMock.expectLastCall();
-        member.ensureActive();
-        PowerMock.expectLastCall();
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
 
-        PowerMock.replayAll();
+        when(statusBackingStore.connectors()).thenReturn(Collections.emptySet());
+        doNothing().when(member).poll(anyLong());
 
         herder.tick();
+
+        // now handle the connector restart
+        doNothing().when(member).ensureActive();
+        when(statusBackingStore.get(CONN1)).thenReturn(null);
+        RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
+        doNothing().when(configBackingStore).putRestartRequest(restartRequest);
+
         FutureCallback<ConnectorStateInfo> callback = new FutureCallback<>();
         herder.restartConnectorAndTasks(restartRequest, callback);
         herder.tick();
         ExecutionException ee = assertThrows(ExecutionException.class, () -> callback.get(1000L, TimeUnit.MILLISECONDS));
         assertTrue(ee.getCause() instanceof NotFoundException);
         assertTrue(ee.getMessage().contains("Status for connector"));
-        PowerMock.verifyAll();
+
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testRestartConnectorAndTasksSuccess() throws Exception {
-        RestartPlan restartPlan = PowerMock.createMock(RestartPlan.class);
-        ConnectorStateInfo connectorStateInfo = PowerMock.createMock(ConnectorStateInfo.class);
-        EasyMock.expect(restartPlan.restartConnectorStateInfo()).andReturn(connectorStateInfo).anyTimes();
-
-        RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
-        EasyMock.expect(herder.buildRestartPlan(restartRequest)).andReturn(Optional.of(restartPlan)).anyTimes();
-
-        configBackingStore.putRestartRequest(restartRequest);
-        PowerMock.expectLastCall();
-
         // get the initial assignment
-        EasyMock.expect(member.memberId()).andStubReturn("leader");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+        when(member.memberId()).thenReturn("leader");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
         expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true);
         expectConfigRefreshAndSnapshot(SNAPSHOT);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
+
+        when(statusBackingStore.connectors()).thenReturn(Collections.emptySet());
+        doNothing().when(member).poll(anyLong());
+
+        herder.tick();
 
         // now handle the connector restart
-        member.wakeup();
-        PowerMock.expectLastCall();
-        member.ensureActive();
-        PowerMock.expectLastCall();
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
+        doNothing().when(member).ensureActive();
 
-        PowerMock.replayAll();
+        RestartPlan restartPlan = mock(RestartPlan.class);
+        ConnectorStateInfo connectorStateInfo = mock(ConnectorStateInfo.class);
+        when(restartPlan.restartConnectorStateInfo()).thenReturn(connectorStateInfo);
+        RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
+        doReturn(Optional.of(restartPlan)).when(herder).buildRestartPlan(restartRequest);
+        doNothing().when(configBackingStore).putRestartRequest(restartRequest);
 
-        herder.tick();
         FutureCallback<ConnectorStateInfo> callback = new FutureCallback<>();
         herder.restartConnectorAndTasks(restartRequest, callback);
         herder.tick();
         assertEquals(connectorStateInfo,  callback.get(1000L, TimeUnit.MILLISECONDS));
-        PowerMock.verifyAll();
+
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testDoRestartConnectorAndTasksEmptyPlan() {
         RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
-        EasyMock.expect(herder.buildRestartPlan(restartRequest)).andReturn(Optional.empty()).anyTimes();
-
-        PowerMock.replayAll();
-
+        doReturn(Optional.empty()).when(herder).buildRestartPlan(restartRequest);
         herder.doRestartConnectorAndTasks(restartRequest);
-        PowerMock.verifyAll();
+
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testDoRestartConnectorAndTasksNoAssignments() {
         ConnectorTaskId taskId = new ConnectorTaskId(CONN1, 0);
         RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
-        RestartPlan restartPlan = PowerMock.createMock(RestartPlan.class);
-        EasyMock.expect(restartPlan.shouldRestartConnector()).andReturn(true).anyTimes();
-        EasyMock.expect(restartPlan.shouldRestartTasks()).andReturn(true).anyTimes();
-        EasyMock.expect(restartPlan.taskIdsToRestart()).andReturn(Collections.singletonList(taskId)).anyTimes();
+        RestartPlan restartPlan = mock(RestartPlan.class);
+        when(restartPlan.shouldRestartConnector()).thenReturn(true);
+        when(restartPlan.taskIdsToRestart()).thenReturn(Collections.singletonList(taskId));
 
-        EasyMock.expect(herder.buildRestartPlan(restartRequest)).andReturn(Optional.of(restartPlan)).anyTimes();
+        doReturn(Optional.of(restartPlan)).when(herder).buildRestartPlan(restartRequest);
 
-        PowerMock.replayAll();
         herder.assignment = ExtendedAssignment.empty();
         herder.doRestartConnectorAndTasks(restartRequest);
-        PowerMock.verifyAll();
+
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testDoRestartConnectorAndTasksOnlyConnector() {
         ConnectorTaskId taskId = new ConnectorTaskId(CONN1, 0);
         RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
-        RestartPlan restartPlan = PowerMock.createMock(RestartPlan.class);
-        EasyMock.expect(restartPlan.shouldRestartConnector()).andReturn(true).anyTimes();
-        EasyMock.expect(restartPlan.shouldRestartTasks()).andReturn(true).anyTimes();
-        EasyMock.expect(restartPlan.taskIdsToRestart()).andReturn(Collections.singletonList(taskId)).anyTimes();
+        RestartPlan restartPlan = mock(RestartPlan.class);
+        when(restartPlan.shouldRestartConnector()).thenReturn(true);
+        when(restartPlan.taskIdsToRestart()).thenReturn(Collections.singletonList(taskId));
 
-        EasyMock.expect(herder.buildRestartPlan(restartRequest)).andReturn(Optional.of(restartPlan)).anyTimes();
+        doReturn(Optional.of(restartPlan)).when(herder).buildRestartPlan(restartRequest);
 
-        herder.assignment = PowerMock.createMock(ExtendedAssignment.class);
-        EasyMock.expect(herder.assignment.connectors()).andReturn(Collections.singletonList(CONN1)).anyTimes();
-        EasyMock.expect(herder.assignment.tasks()).andReturn(Collections.emptyList()).anyTimes();
+        herder.assignment = mock(ExtendedAssignment.class);
+        when(herder.assignment.connectors()).thenReturn(Collections.singletonList(CONN1));
+        when(herder.assignment.tasks()).thenReturn(Collections.emptyList());
 
-        worker.stopAndAwaitConnector(CONN1);
-        PowerMock.expectLastCall();
+        doNothing().when(worker).stopAndAwaitConnector(CONN1);
 
-        Capture<Callback<TargetState>>  stateCallback = newCapture();
-        worker.startConnector(EasyMock.eq(CONN1), EasyMock.anyObject(), EasyMock.anyObject(),
-                EasyMock.eq(herder), EasyMock.anyObject(TargetState.class), capture(stateCallback));
-        PowerMock.expectLastCall().andAnswer(() -> {
+        ArgumentCaptor<Callback<TargetState>>  stateCallback = ArgumentCaptor.forClass(Callback.class);
+        doAnswer(invocation -> {
             stateCallback.getValue().onCompletion(null, TargetState.STARTED);
             return true;
-        });
-        member.wakeup();
-        PowerMock.expectLastCall();
+        }).when(worker).startConnector(eq(CONN1), any(), any(), eq(herder), any(), stateCallback.capture());
+        doNothing().when(member).wakeup();
+        doNothing().when(statusBackingStore).put(any(ConnectorStatus.class));
 
-
-        herder.onRestart(CONN1);
-        EasyMock.expectLastCall();
-
-        PowerMock.replayAll();
         herder.doRestartConnectorAndTasks(restartRequest);
-        PowerMock.verifyAll();
+
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testDoRestartConnectorAndTasksOnlyTasks() {
         RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
-        RestartPlan restartPlan = PowerMock.createMock(RestartPlan.class);
-        EasyMock.expect(restartPlan.shouldRestartConnector()).andReturn(true).anyTimes();
-        EasyMock.expect(restartPlan.shouldRestartTasks()).andReturn(true).anyTimes();
+        RestartPlan restartPlan = mock(RestartPlan.class);
+        when(restartPlan.shouldRestartConnector()).thenReturn(true);
         // The connector has three tasks
-        EasyMock.expect(restartPlan.taskIdsToRestart()).andReturn(Arrays.asList(TASK0, TASK1, TASK2)).anyTimes();
-        EasyMock.expect(restartPlan.restartTaskCount()).andReturn(3).anyTimes();
-        EasyMock.expect(restartPlan.totalTaskCount()).andReturn(3).anyTimes();
-        EasyMock.expect(herder.buildRestartPlan(restartRequest)).andReturn(Optional.of(restartPlan)).anyTimes();
+        when(restartPlan.taskIdsToRestart()).thenReturn(Arrays.asList(TASK0, TASK1, TASK2));
+        when(restartPlan.totalTaskCount()).thenReturn(3);
+        doReturn(Optional.of(restartPlan)).when(herder).buildRestartPlan(restartRequest);
 
-        herder.assignment = PowerMock.createMock(ExtendedAssignment.class);
-        EasyMock.expect(herder.assignment.connectors()).andReturn(Collections.emptyList()).anyTimes();
+        herder.assignment = mock(ExtendedAssignment.class);
+        when(herder.assignment.connectors()).thenReturn(Collections.emptyList());
         // But only one task is assigned to this worker
-        EasyMock.expect(herder.assignment.tasks()).andReturn(Collections.singletonList(TASK0)).anyTimes();
+        when(herder.assignment.tasks()).thenReturn(Collections.singletonList(TASK0));
 
         herder.configState = SNAPSHOT;
 
-        worker.stopAndAwaitTasks(Collections.singletonList(TASK0));
-        PowerMock.expectLastCall();
-
-        herder.onRestart(TASK0);
-        EasyMock.expectLastCall();
-
-        worker.startSourceTask(EasyMock.eq(TASK0), EasyMock.anyObject(), EasyMock.anyObject(), EasyMock.anyObject(),
-                EasyMock.eq(herder), EasyMock.anyObject(TargetState.class));
-        PowerMock.expectLastCall().andReturn(true);
+        doNothing().when(worker).stopAndAwaitTasks(Collections.singletonList(TASK0));
+        doNothing().when(statusBackingStore).put(any(TaskStatus.class));
+        when(worker.startSourceTask(eq(TASK0), any(), any(), any(), eq(herder), any())).thenReturn(true);
 
-        PowerMock.replayAll();
         herder.doRestartConnectorAndTasks(restartRequest);
-        PowerMock.verifyAll();
+
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testDoRestartConnectorAndTasksBoth() {
         ConnectorTaskId taskId = new ConnectorTaskId(CONN1, 0);
         RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
-        RestartPlan restartPlan = PowerMock.createMock(RestartPlan.class);
-        EasyMock.expect(restartPlan.shouldRestartConnector()).andReturn(true).anyTimes();
-        EasyMock.expect(restartPlan.shouldRestartTasks()).andReturn(true).anyTimes();
-        EasyMock.expect(restartPlan.taskIdsToRestart()).andReturn(Collections.singletonList(taskId)).anyTimes();
-        EasyMock.expect(restartPlan.restartTaskCount()).andReturn(1).anyTimes();
-        EasyMock.expect(restartPlan.totalTaskCount()).andReturn(1).anyTimes();
-        EasyMock.expect(herder.buildRestartPlan(restartRequest)).andReturn(Optional.of(restartPlan)).anyTimes();
-
-        herder.assignment = PowerMock.createMock(ExtendedAssignment.class);
-        EasyMock.expect(herder.assignment.connectors()).andReturn(Collections.singletonList(CONN1)).anyTimes();
-        EasyMock.expect(herder.assignment.tasks()).andReturn(Collections.singletonList(taskId)).anyTimes();
-
-        herder.configState = SNAPSHOT;
+        RestartPlan restartPlan = mock(RestartPlan.class);
+        when(restartPlan.shouldRestartConnector()).thenReturn(true);
+        when(restartPlan.taskIdsToRestart()).thenReturn(Collections.singletonList(taskId));
+        when(restartPlan.totalTaskCount()).thenReturn(1);
+        doReturn(Optional.of(restartPlan)).when(herder).buildRestartPlan(restartRequest);
 
-        worker.stopAndAwaitConnector(CONN1);
-        PowerMock.expectLastCall();
-
-        Capture<Callback<TargetState>>  stateCallback = newCapture();
-        worker.startConnector(EasyMock.eq(CONN1), EasyMock.anyObject(), EasyMock.anyObject(),
-                EasyMock.eq(herder), EasyMock.anyObject(TargetState.class), capture(stateCallback));
+        herder.assignment = mock(ExtendedAssignment.class);
+        when(herder.assignment.connectors()).thenReturn(Collections.singletonList(CONN1));
+        when(herder.assignment.tasks()).thenReturn(Collections.singletonList(taskId));
 
+        herder.configState = SNAPSHOT;
 
-        herder.onRestart(CONN1);
-        EasyMock.expectLastCall();
+        doNothing().when(worker).stopAndAwaitConnector(CONN1);
 
-        worker.stopAndAwaitTasks(Collections.singletonList(taskId));
-        PowerMock.expectLastCall();
+        ArgumentCaptor<Callback<TargetState>>  stateCallback = ArgumentCaptor.forClass(Callback.class);
+        doAnswer(invocation -> {
+            stateCallback.getValue().onCompletion(null, TargetState.STARTED);
+            return true;
+        }).when(worker).startConnector(eq(CONN1), any(), any(), eq(herder), any(), stateCallback.capture());
+        doNothing().when(member).wakeup();
+        doNothing().when(statusBackingStore).put(any(ConnectorStatus.class));
 
-        herder.onRestart(taskId);
-        EasyMock.expectLastCall();
+        doNothing().when(worker).stopAndAwaitTasks(Collections.singletonList(taskId));
 
-        worker.startSourceTask(EasyMock.eq(TASK0), EasyMock.anyObject(), EasyMock.anyObject(), EasyMock.anyObject(),
-                EasyMock.eq(herder), EasyMock.anyObject(TargetState.class));
-        PowerMock.expectLastCall().andReturn(true);
+        when(worker.startSourceTask(eq(TASK0), any(), any(), any(), eq(herder), any())).thenReturn(true);
+        doNothing().when(statusBackingStore).put(any(TaskStatus.class));
 
-        PowerMock.replayAll();
         herder.doRestartConnectorAndTasks(restartRequest);
-        PowerMock.verifyAll();
+
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testRestartTask() throws Exception {
-        EasyMock.expect(worker.connectorTaskConfigs(CONN1, conn1SinkConfig)).andStubReturn(TASK_CONFIGS);
-
         // get the initial assignment
-        EasyMock.expect(member.memberId()).andStubReturn("leader");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+        when(member.memberId()).thenReturn("leader");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
         expectRebalance(1, Collections.emptyList(), singletonList(TASK0), true);
         expectConfigRefreshAndSnapshot(SNAPSHOT);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
-        worker.startSourceTask(EasyMock.eq(TASK0), EasyMock.anyObject(), EasyMock.anyObject(), EasyMock.anyObject(),
-                EasyMock.eq(herder), EasyMock.eq(TargetState.STARTED));
-        PowerMock.expectLastCall().andReturn(true);
-
-        // now handle the task restart
-        member.wakeup();
-        PowerMock.expectLastCall();
-        member.ensureActive();
-        PowerMock.expectLastCall();
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
 
-        worker.stopAndAwaitTask(TASK0);
-        PowerMock.expectLastCall();
-        worker.startSourceTask(EasyMock.eq(TASK0), EasyMock.anyObject(), EasyMock.anyObject(), EasyMock.anyObject(),
-                EasyMock.eq(herder), EasyMock.eq(TargetState.STARTED));
-        PowerMock.expectLastCall().andReturn(true);
+        when(statusBackingStore.connectors()).thenReturn(Collections.emptySet());
+        doNothing().when(member).poll(anyLong());
 
-        PowerMock.replayAll();
+        when(worker.startSourceTask(eq(TASK0), any(), any(), any(), eq(herder), any())).thenReturn(true);
 
         herder.tick();
+
+        // now handle the task restart
+        doNothing().when(member).ensureActive();
+        doNothing().when(worker).stopAndAwaitTask(TASK0);
         FutureCallback<Void> callback = new FutureCallback<>();
         herder.restartTask(TASK0, callback);
         herder.tick();
         callback.get(1000L, TimeUnit.MILLISECONDS);
 
-        PowerMock.verifyAll();
+        verify(worker, times(2)).startSourceTask(eq(TASK0), any(), any(), any(), eq(herder), any());
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testRestartUnknownTask() throws Exception {
         // get the initial assignment
-        EasyMock.expect(member.memberId()).andStubReturn("member");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+        when(member.memberId()).thenReturn("member");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
         expectRebalance(1, Collections.emptyList(), Collections.emptyList());
         expectConfigRefreshAndSnapshot(SNAPSHOT);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
-
-        member.wakeup();
-        PowerMock.expectLastCall();
-        member.ensureActive();
-        PowerMock.expectLastCall();
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
+        doNothing().when(member).poll(anyLong());
 
-        PowerMock.replayAll();
+        herder.tick();
 
+        doNothing().when(member).ensureActive();
         FutureCallback<Void> callback = new FutureCallback<>();
-        herder.tick();
         herder.restartTask(new ConnectorTaskId("blah", 0), callback);
         herder.tick();
 
-        try {
-            callback.get(1000L, TimeUnit.MILLISECONDS);
-            fail("Expected NotLeaderException to be raised");
-        } catch (ExecutionException e) {
-            assertTrue(e.getCause() instanceof NotFoundException);
-        }
+        ExecutionException e = assertThrows(ExecutionException.class, () -> callback.get(1000, TimeUnit.MILLISECONDS));
+        assertEquals(NotFoundException.class, e.getCause().getClass());
 
-        PowerMock.verifyAll();
-    }
-
-    @Test
-    public void testRequestProcessingOrder() {
-        final DistributedHerder.DistributedHerderRequest req1 = herder.addRequest(100, null, null);
-        final DistributedHerder.DistributedHerderRequest req2 = herder.addRequest(10, null, null);
-        final DistributedHerder.DistributedHerderRequest req3 = herder.addRequest(200, null, null);
-        final DistributedHerder.DistributedHerderRequest req4 = herder.addRequest(200, null, null);
-
-        assertEquals(req2, herder.requests.pollFirst()); // lowest delay
-        assertEquals(req1, herder.requests.pollFirst()); // next lowest delay
-        assertEquals(req3, herder.requests.pollFirst()); // same delay as req4, but added first
-        assertEquals(req4, herder.requests.pollFirst());
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testRestartTaskRedirectToLeader() throws Exception {
         // get the initial assignment
-        EasyMock.expect(member.memberId()).andStubReturn("member");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+        when(member.memberId()).thenReturn("member");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
         expectRebalance(1, Collections.emptyList(), Collections.emptyList());
         expectConfigRefreshAndSnapshot(SNAPSHOT);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
-
-        // now handle the task restart
-        member.wakeup();
-        PowerMock.expectLastCall();
-        member.ensureActive();
-        PowerMock.expectLastCall();
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
-
-        PowerMock.replayAll();
+        doNothing().when(member).poll(anyLong());
 
         herder.tick();
+
+        // now handle the task restart
+        doNothing().when(member).ensureActive();
         FutureCallback<Void> callback = new FutureCallback<>();
         herder.restartTask(TASK0, callback);
         herder.tick();
 
-        try {
-            callback.get(1000L, TimeUnit.MILLISECONDS);
-            fail("Expected NotLeaderException to be raised");
-        } catch (ExecutionException e) {
-            assertTrue(e.getCause() instanceof NotLeaderException);
-        }
+        ExecutionException e = assertThrows(ExecutionException.class, () -> callback.get(1000, TimeUnit.MILLISECONDS));
+        assertEquals(NotLeaderException.class, e.getCause().getClass());
 
-        PowerMock.verifyAll();
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testRestartTaskRedirectToOwner() throws Exception {
         // get the initial assignment
-        EasyMock.expect(member.memberId()).andStubReturn("leader");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+        when(member.memberId()).thenReturn("leader");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
         expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true);
         expectConfigRefreshAndSnapshot(SNAPSHOT);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
+
+        when(statusBackingStore.connectors()).thenReturn(Collections.emptySet());
+        doNothing().when(member).poll(anyLong());
+
+        herder.tick();
 
         // now handle the task restart
         String ownerUrl = "ownerUrl";
-        EasyMock.expect(member.ownerUrl(TASK0)).andReturn(ownerUrl);
-        member.wakeup();
-        PowerMock.expectLastCall();
-        member.ensureActive();
-        PowerMock.expectLastCall();
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
+        when(member.ownerUrl(TASK0)).thenReturn(ownerUrl);
+        doNothing().when(member).ensureActive();
 
-        PowerMock.replayAll();
-
-        herder.tick();
         FutureCallback<Void> callback = new FutureCallback<>();
         herder.restartTask(TASK0, callback);
         herder.tick();
 
-        try {
-            callback.get(1000L, TimeUnit.MILLISECONDS);
-            fail("Expected NotLeaderException to be raised");
-        } catch (ExecutionException e) {
-            assertTrue(e.getCause() instanceof NotAssignedException);
-            NotAssignedException notAssignedException = (NotAssignedException) e.getCause();
-            assertEquals(ownerUrl, notAssignedException.forwardUrl());
-        }
+        ExecutionException e = assertThrows(ExecutionException.class, () -> callback.get(1000, TimeUnit.MILLISECONDS));
+        assertEquals(NotAssignedException.class, e.getCause().getClass());
+        assertEquals(ownerUrl, ((NotAssignedException) e.getCause()).forwardUrl());
 
-        PowerMock.verifyAll();
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
-    public void testConnectorConfigAdded() {
+    public void testRequestProcessingOrder() {
+        final DistributedHerder.DistributedHerderRequest req1 = herder.addRequest(100, null, null);
+        final DistributedHerder.DistributedHerderRequest req2 = herder.addRequest(10, null, null);
+        final DistributedHerder.DistributedHerderRequest req3 = herder.addRequest(200, null, null);
+        final DistributedHerder.DistributedHerderRequest req4 = herder.addRequest(200, null, null);
+
+        assertEquals(req2, herder.requests.pollFirst()); // lowest delay
+        assertEquals(req1, herder.requests.pollFirst()); // next lowest delay
+        assertEquals(req3, herder.requests.pollFirst()); // same delay as req4, but added first
+        assertEquals(req4, herder.requests.pollFirst());
+    }
+
+    @Test
+    public void testConnectorConfigAdded() throws Exception {
         // If a connector was added, we need to rebalance
-        EasyMock.expect(member.memberId()).andStubReturn("member");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+        when(member.memberId()).thenReturn("member");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
 
         // join, no configs so no need to catch up on config topic
         expectRebalance(-1, Collections.emptyList(), Collections.emptyList());
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
+        doNothing().when(member).poll(anyLong());
+
+        herder.tick(); // join
 
-        // apply config
-        member.wakeup();
-        PowerMock.expectLastCall();
-        member.ensureActive();
-        PowerMock.expectLastCall();
         // Checks for config updates and starts rebalance
-        EasyMock.expect(configBackingStore.snapshot()).andReturn(SNAPSHOT);
-        member.requestRejoin();
-        PowerMock.expectLastCall();
+        doNothing().when(configBackingStore).refresh(anyLong(), any());
+        when(configBackingStore.snapshot()).thenReturn(SNAPSHOT);

Review Comment:
   Whoops 🤦 



##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java:
##########
@@ -3724,62 +2818,54 @@ public void testExternalZombieFencingRequestDelayedCompletion() throws Exception
                 pendingFencing,
                 tasksPerConnector
         );
-        tasksPerConnector.keySet().forEach(c -> expectConfigRefreshAndSnapshot(configState));
+        expectConfigRefreshAndSnapshot(configState);
+
+        when(statusBackingStore.connectors()).thenReturn(Collections.emptySet());
+        doNothing().when(member).poll(anyLong());
 
         // The callbacks that the herder has accrued for outstanding fencing futures, which will be completed after
         // a successful round of fencing and a task record write to the config topic
-        Map<String, Capture<KafkaFuture.BiConsumer<Void, Throwable>>> herderFencingCallbacks = new HashMap<>();
+        Map<String, ArgumentCaptor<KafkaFuture.BiConsumer<Void, Throwable>>> herderFencingCallbacks = new HashMap<>();
         // The callbacks that the herder has installed for after a successful round of zombie fencing, but before writing
         // a task record to the config topic
-        Map<String, Capture<KafkaFuture.BaseFunction<Void, Void>>> workerFencingFollowups = new HashMap<>();
+        Map<String, ArgumentCaptor<KafkaFuture.BaseFunction<Void, Void>>> workerFencingFollowups = new HashMap<>();
 
         Map<String, CountDownLatch> callbacksInstalled = new HashMap<>();
         tasksPerConnector.forEach((connector, numStackedRequests) -> {
             // The future returned by Worker::fenceZombies
-            KafkaFuture<Void> workerFencingFuture = EasyMock.mock(KafkaFuture.class);
-            // The future tracked by the herder (which tracks the fencing performed by the worker and the possible followup write to the config topic) 
-            KafkaFuture<Void> herderFencingFuture = EasyMock.mock(KafkaFuture.class);
+            KafkaFuture<Void> workerFencingFuture = mock(KafkaFuture.class);
+            // The future tracked by the herder (which tracks the fencing performed by the worker and the possible followup write to the config topic)
+            KafkaFuture<Void> herderFencingFuture = mock(KafkaFuture.class);
 
-            Capture<KafkaFuture.BiConsumer<Void, Throwable>> herderFencingCallback = EasyMock.newCapture(CaptureType.ALL);
+            ArgumentCaptor<KafkaFuture.BiConsumer<Void, Throwable>> herderFencingCallback = ArgumentCaptor.forClass(KafkaFuture.BiConsumer.class);
             herderFencingCallbacks.put(connector, herderFencingCallback);
 
             // Don't immediately invoke callbacks that the herder sets up for when the worker fencing and writes to the config topic have completed
             // Instead, wait for them to be installed, then invoke them explicitly after the fact on a thread separate from the herder's tick thread
-            EasyMock.expect(herderFencingFuture.whenComplete(EasyMock.capture(herderFencingCallback)))
-                    .andReturn(null)
-                    .times(numStackedRequests + 1);
+            when(herderFencingFuture.whenComplete(herderFencingCallback.capture())).thenReturn(null);
 
-            Capture<KafkaFuture.BaseFunction<Void, Void>> fencingFollowup = EasyMock.newCapture();
+            ArgumentCaptor<KafkaFuture.BaseFunction<Void, Void>> fencingFollowup = ArgumentCaptor.forClass(KafkaFuture.BaseFunction.class);
             CountDownLatch callbackInstalled = new CountDownLatch(1);
             workerFencingFollowups.put(connector, fencingFollowup);
             callbacksInstalled.put(connector, callbackInstalled);
-            EasyMock.expect(workerFencingFuture.thenApply(EasyMock.capture(fencingFollowup))).andAnswer(() -> {
+            when(workerFencingFuture.thenApply(fencingFollowup.capture())).thenAnswer(invocation -> {
                 callbackInstalled.countDown();
                 return herderFencingFuture;
             });
 
             // We should only perform a single physical zombie fencing; all the subsequent requests should be stacked onto the first one
-            EasyMock.expect(worker.fenceZombies(
-                    EasyMock.eq(connector), EasyMock.eq(taskCountRecords.get(connector)), EasyMock.anyObject())
-            ).andReturn(workerFencingFuture);
-
-            for (int i = 0; i < numStackedRequests; i++) {
-                expectConfigRefreshAndSnapshot(configState);
-            }
-
-            PowerMock.replay(workerFencingFuture, herderFencingFuture);
+            when(worker.fenceZombies(eq(connector), eq(taskCountRecords.get(connector)), any()))
+                    .thenReturn(workerFencingFuture)
+                    .thenAnswer(invocation -> {
+                        fail("Expected only a single zombie fencing per connector");

Review Comment:
   Oh, that's a good point, it would not be bubbled up correctly in fact.



##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java:
##########
@@ -1466,560 +1032,421 @@ public void testRestartConnectorAndTasksUnknownConnector() throws Exception {
         assertTrue(ee.getCause() instanceof NotFoundException);
         assertTrue(ee.getMessage().contains("Unknown connector:"));
 
-        PowerMock.verifyAll();
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testRestartConnectorAndTasksNotLeader() throws Exception {
-        RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
-
         // get the initial assignment
-        EasyMock.expect(member.memberId()).andStubReturn("member");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+        when(member.memberId()).thenReturn("member");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
         expectRebalance(1, Collections.emptyList(), Collections.emptyList());
         expectConfigRefreshAndSnapshot(SNAPSHOT);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
+        doNothing().when(member).poll(anyLong());
 
-        // now handle the connector restart
-        member.wakeup();
-        PowerMock.expectLastCall();
-        member.ensureActive();
-        PowerMock.expectLastCall();
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
+        herder.tick();
 
-        PowerMock.replayAll();
+        // now handle the connector restart
+        doNothing().when(member).ensureActive();
 
         herder.tick();
         FutureCallback<ConnectorStateInfo> callback = new FutureCallback<>();
+        RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
         herder.restartConnectorAndTasks(restartRequest, callback);
         herder.tick();
         ExecutionException ee = assertThrows(ExecutionException.class, () -> callback.get(1000L, TimeUnit.MILLISECONDS));
         assertTrue(ee.getCause() instanceof NotLeaderException);
 
-        PowerMock.verifyAll();
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testRestartConnectorAndTasksUnknownStatus() throws Exception {
-        RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
-        EasyMock.expect(herder.buildRestartPlan(restartRequest)).andReturn(Optional.empty()).anyTimes();
-
-        configBackingStore.putRestartRequest(restartRequest);
-        PowerMock.expectLastCall();
-
         // get the initial assignment
-        EasyMock.expect(member.memberId()).andStubReturn("leader");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+        when(member.memberId()).thenReturn("leader");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
         expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true);
         expectConfigRefreshAndSnapshot(SNAPSHOT);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
-
-        // now handle the connector restart
-        member.wakeup();
-        PowerMock.expectLastCall();
-        member.ensureActive();
-        PowerMock.expectLastCall();
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
 
-        PowerMock.replayAll();
+        when(statusBackingStore.connectors()).thenReturn(Collections.emptySet());
+        doNothing().when(member).poll(anyLong());
 
         herder.tick();
+
+        // now handle the connector restart
+        doNothing().when(member).ensureActive();
+        when(statusBackingStore.get(CONN1)).thenReturn(null);
+        RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
+        doNothing().when(configBackingStore).putRestartRequest(restartRequest);
+
         FutureCallback<ConnectorStateInfo> callback = new FutureCallback<>();
         herder.restartConnectorAndTasks(restartRequest, callback);
         herder.tick();
         ExecutionException ee = assertThrows(ExecutionException.class, () -> callback.get(1000L, TimeUnit.MILLISECONDS));
         assertTrue(ee.getCause() instanceof NotFoundException);
         assertTrue(ee.getMessage().contains("Status for connector"));
-        PowerMock.verifyAll();
+
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testRestartConnectorAndTasksSuccess() throws Exception {
-        RestartPlan restartPlan = PowerMock.createMock(RestartPlan.class);
-        ConnectorStateInfo connectorStateInfo = PowerMock.createMock(ConnectorStateInfo.class);
-        EasyMock.expect(restartPlan.restartConnectorStateInfo()).andReturn(connectorStateInfo).anyTimes();
-
-        RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
-        EasyMock.expect(herder.buildRestartPlan(restartRequest)).andReturn(Optional.of(restartPlan)).anyTimes();
-
-        configBackingStore.putRestartRequest(restartRequest);
-        PowerMock.expectLastCall();
-
         // get the initial assignment
-        EasyMock.expect(member.memberId()).andStubReturn("leader");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+        when(member.memberId()).thenReturn("leader");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
         expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true);
         expectConfigRefreshAndSnapshot(SNAPSHOT);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
+
+        when(statusBackingStore.connectors()).thenReturn(Collections.emptySet());
+        doNothing().when(member).poll(anyLong());
+
+        herder.tick();
 
         // now handle the connector restart
-        member.wakeup();
-        PowerMock.expectLastCall();
-        member.ensureActive();
-        PowerMock.expectLastCall();
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
+        doNothing().when(member).ensureActive();
 
-        PowerMock.replayAll();
+        RestartPlan restartPlan = mock(RestartPlan.class);
+        ConnectorStateInfo connectorStateInfo = mock(ConnectorStateInfo.class);
+        when(restartPlan.restartConnectorStateInfo()).thenReturn(connectorStateInfo);
+        RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
+        doReturn(Optional.of(restartPlan)).when(herder).buildRestartPlan(restartRequest);
+        doNothing().when(configBackingStore).putRestartRequest(restartRequest);
 
-        herder.tick();
         FutureCallback<ConnectorStateInfo> callback = new FutureCallback<>();
         herder.restartConnectorAndTasks(restartRequest, callback);
         herder.tick();
         assertEquals(connectorStateInfo,  callback.get(1000L, TimeUnit.MILLISECONDS));
-        PowerMock.verifyAll();
+
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testDoRestartConnectorAndTasksEmptyPlan() {
         RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
-        EasyMock.expect(herder.buildRestartPlan(restartRequest)).andReturn(Optional.empty()).anyTimes();
-
-        PowerMock.replayAll();
-
+        doReturn(Optional.empty()).when(herder).buildRestartPlan(restartRequest);
         herder.doRestartConnectorAndTasks(restartRequest);
-        PowerMock.verifyAll();
+
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testDoRestartConnectorAndTasksNoAssignments() {
         ConnectorTaskId taskId = new ConnectorTaskId(CONN1, 0);
         RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
-        RestartPlan restartPlan = PowerMock.createMock(RestartPlan.class);
-        EasyMock.expect(restartPlan.shouldRestartConnector()).andReturn(true).anyTimes();
-        EasyMock.expect(restartPlan.shouldRestartTasks()).andReturn(true).anyTimes();
-        EasyMock.expect(restartPlan.taskIdsToRestart()).andReturn(Collections.singletonList(taskId)).anyTimes();
+        RestartPlan restartPlan = mock(RestartPlan.class);
+        when(restartPlan.shouldRestartConnector()).thenReturn(true);
+        when(restartPlan.taskIdsToRestart()).thenReturn(Collections.singletonList(taskId));
 
-        EasyMock.expect(herder.buildRestartPlan(restartRequest)).andReturn(Optional.of(restartPlan)).anyTimes();
+        doReturn(Optional.of(restartPlan)).when(herder).buildRestartPlan(restartRequest);
 
-        PowerMock.replayAll();
         herder.assignment = ExtendedAssignment.empty();
         herder.doRestartConnectorAndTasks(restartRequest);
-        PowerMock.verifyAll();
+
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testDoRestartConnectorAndTasksOnlyConnector() {
         ConnectorTaskId taskId = new ConnectorTaskId(CONN1, 0);
         RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
-        RestartPlan restartPlan = PowerMock.createMock(RestartPlan.class);
-        EasyMock.expect(restartPlan.shouldRestartConnector()).andReturn(true).anyTimes();
-        EasyMock.expect(restartPlan.shouldRestartTasks()).andReturn(true).anyTimes();
-        EasyMock.expect(restartPlan.taskIdsToRestart()).andReturn(Collections.singletonList(taskId)).anyTimes();
+        RestartPlan restartPlan = mock(RestartPlan.class);
+        when(restartPlan.shouldRestartConnector()).thenReturn(true);
+        when(restartPlan.taskIdsToRestart()).thenReturn(Collections.singletonList(taskId));
 
-        EasyMock.expect(herder.buildRestartPlan(restartRequest)).andReturn(Optional.of(restartPlan)).anyTimes();
+        doReturn(Optional.of(restartPlan)).when(herder).buildRestartPlan(restartRequest);
 
-        herder.assignment = PowerMock.createMock(ExtendedAssignment.class);
-        EasyMock.expect(herder.assignment.connectors()).andReturn(Collections.singletonList(CONN1)).anyTimes();
-        EasyMock.expect(herder.assignment.tasks()).andReturn(Collections.emptyList()).anyTimes();
+        herder.assignment = mock(ExtendedAssignment.class);
+        when(herder.assignment.connectors()).thenReturn(Collections.singletonList(CONN1));
+        when(herder.assignment.tasks()).thenReturn(Collections.emptyList());
 
-        worker.stopAndAwaitConnector(CONN1);
-        PowerMock.expectLastCall();
+        doNothing().when(worker).stopAndAwaitConnector(CONN1);
 
-        Capture<Callback<TargetState>>  stateCallback = newCapture();
-        worker.startConnector(EasyMock.eq(CONN1), EasyMock.anyObject(), EasyMock.anyObject(),
-                EasyMock.eq(herder), EasyMock.anyObject(TargetState.class), capture(stateCallback));
-        PowerMock.expectLastCall().andAnswer(() -> {
+        ArgumentCaptor<Callback<TargetState>>  stateCallback = ArgumentCaptor.forClass(Callback.class);
+        doAnswer(invocation -> {
             stateCallback.getValue().onCompletion(null, TargetState.STARTED);
             return true;
-        });
-        member.wakeup();
-        PowerMock.expectLastCall();
+        }).when(worker).startConnector(eq(CONN1), any(), any(), eq(herder), any(), stateCallback.capture());
+        doNothing().when(member).wakeup();
+        doNothing().when(statusBackingStore).put(any(ConnectorStatus.class));

Review Comment:
   > One could be to set up a mock expectation with the exact status we expect to see reported to the status store
   
   Makes sense, I've made this change here and in the subsequent tests as well. I guess I just put in the lowest impedance change initially since it's additional coverage (but was required to satisfy the new stricter verifications) 😅 



##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java:
##########
@@ -3003,41 +2192,39 @@ public void testKeyRotationWhenWorkerBecomesLeader() throws Exception {
                 Collections.emptySet(),
                 Collections.emptySet());
         expectConfigRefreshAndSnapshot(snapshotWithKey);
-        // Second rebalance: poll indefinitely as worker is follower, so expiration still doesn't come into play
-        member.poll(Long.MAX_VALUE);
-        EasyMock.expectLastCall();
+
+        configUpdateListener.onSessionKeyUpdate(initialKey);
+        herder.tick();
 
         expectRebalance(2, Collections.emptyList(), Collections.emptyList(), "member", MEMBER_URL, true);
-        Capture<SessionKey> updatedKey = EasyMock.newCapture();
-        configBackingStore.putSessionKey(EasyMock.capture(updatedKey));
-        EasyMock.expectLastCall().andAnswer(() -> {
+        when(statusBackingStore.connectors()).thenReturn(Collections.emptySet());
+        ArgumentCaptor<SessionKey> updatedKey = ArgumentCaptor.forClass(SessionKey.class);
+        doAnswer(invocation -> {
             configUpdateListener.onSessionKeyUpdate(updatedKey.getValue());
             return null;
-        });
-        // Third rebalance: poll for a limited time as worker has become leader and must wake up for key expiration
-        member.poll(leq(rotationTtlDelay));
-        EasyMock.expectLastCall();
+        }).when(configBackingStore).putSessionKey(updatedKey.capture());
 
-        PowerMock.replayAll();
-
-        herder.tick();
-        configUpdateListener.onSessionKeyUpdate(initialKey);
-        herder.tick();
         herder.tick();
 
-        PowerMock.verifyAll();
+        // First rebalance: poll indefinitely as no key has been read yet, so expiration doesn't come into play
+        // Second rebalance: poll indefinitely as worker is follower, so expiration still doesn't come into play
+        verify(member, times(2)).poll(eq(Long.MAX_VALUE));

Review Comment:
   Good idea, done 



##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java:
##########
@@ -3872,7 +2952,7 @@ public void testVerifyTaskGeneration() {
         assertThrows(ConnectException.class, () -> herder.verifyTaskGenerationAndOwnership(unassignedTask, 1, verifyCallback));
         assertThrows(ConnectException.class, () -> herder.verifyTaskGenerationAndOwnership(unassignedTask, 2, verifyCallback));
 
-        PowerMock.verifyAll();
+        verify(verifyCallback, times(3)).onCompletion(isNull(), isNull());

Review Comment:
   > We expect this to be invoked five times in the original test
   
   There are only 3 calls to `herder::verifyTaskGenerationAndOwnership` in this test that aren't expected to throw an exception. 
   
   > Were we never properly verifying the number of invocations in the existing test?
   
   Yep, the `verifyCallback` mock isn't passed into `PowerMock::replayAll`  which means that it isn't "known" to `PowerMock` and the `PowerMock::verifyAll` will not make any verifications on it. If this line - 
   https://github.com/apache/kafka/blob/f23394336a7741bf4eb23fcde951af0a23a69bd0/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java#L3838
   
   is changed to `PowerMock.replayAll(verifyCallback);` then the test fails and only passes when we change this loop from 5 iterations to 3 -
   https://github.com/apache/kafka/blob/f23394336a7741bf4eb23fcde951af0a23a69bd0/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java#L3833-L3836



##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java:
##########
@@ -1466,560 +1032,421 @@ public void testRestartConnectorAndTasksUnknownConnector() throws Exception {
         assertTrue(ee.getCause() instanceof NotFoundException);
         assertTrue(ee.getMessage().contains("Unknown connector:"));
 
-        PowerMock.verifyAll();
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testRestartConnectorAndTasksNotLeader() throws Exception {
-        RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
-
         // get the initial assignment
-        EasyMock.expect(member.memberId()).andStubReturn("member");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+        when(member.memberId()).thenReturn("member");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
         expectRebalance(1, Collections.emptyList(), Collections.emptyList());
         expectConfigRefreshAndSnapshot(SNAPSHOT);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
+        doNothing().when(member).poll(anyLong());
 
-        // now handle the connector restart
-        member.wakeup();
-        PowerMock.expectLastCall();
-        member.ensureActive();
-        PowerMock.expectLastCall();
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
+        herder.tick();
 
-        PowerMock.replayAll();
+        // now handle the connector restart
+        doNothing().when(member).ensureActive();
 
         herder.tick();
         FutureCallback<ConnectorStateInfo> callback = new FutureCallback<>();
+        RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
         herder.restartConnectorAndTasks(restartRequest, callback);
         herder.tick();
         ExecutionException ee = assertThrows(ExecutionException.class, () -> callback.get(1000L, TimeUnit.MILLISECONDS));
         assertTrue(ee.getCause() instanceof NotLeaderException);
 
-        PowerMock.verifyAll();
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testRestartConnectorAndTasksUnknownStatus() throws Exception {
-        RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
-        EasyMock.expect(herder.buildRestartPlan(restartRequest)).andReturn(Optional.empty()).anyTimes();
-
-        configBackingStore.putRestartRequest(restartRequest);
-        PowerMock.expectLastCall();
-
         // get the initial assignment
-        EasyMock.expect(member.memberId()).andStubReturn("leader");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+        when(member.memberId()).thenReturn("leader");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
         expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true);
         expectConfigRefreshAndSnapshot(SNAPSHOT);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
-
-        // now handle the connector restart
-        member.wakeup();
-        PowerMock.expectLastCall();
-        member.ensureActive();
-        PowerMock.expectLastCall();
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
 
-        PowerMock.replayAll();
+        when(statusBackingStore.connectors()).thenReturn(Collections.emptySet());
+        doNothing().when(member).poll(anyLong());
 
         herder.tick();
+
+        // now handle the connector restart
+        doNothing().when(member).ensureActive();
+        when(statusBackingStore.get(CONN1)).thenReturn(null);
+        RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
+        doNothing().when(configBackingStore).putRestartRequest(restartRequest);
+
         FutureCallback<ConnectorStateInfo> callback = new FutureCallback<>();
         herder.restartConnectorAndTasks(restartRequest, callback);
         herder.tick();
         ExecutionException ee = assertThrows(ExecutionException.class, () -> callback.get(1000L, TimeUnit.MILLISECONDS));
         assertTrue(ee.getCause() instanceof NotFoundException);
         assertTrue(ee.getMessage().contains("Status for connector"));
-        PowerMock.verifyAll();
+
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testRestartConnectorAndTasksSuccess() throws Exception {
-        RestartPlan restartPlan = PowerMock.createMock(RestartPlan.class);
-        ConnectorStateInfo connectorStateInfo = PowerMock.createMock(ConnectorStateInfo.class);
-        EasyMock.expect(restartPlan.restartConnectorStateInfo()).andReturn(connectorStateInfo).anyTimes();
-
-        RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
-        EasyMock.expect(herder.buildRestartPlan(restartRequest)).andReturn(Optional.of(restartPlan)).anyTimes();
-
-        configBackingStore.putRestartRequest(restartRequest);
-        PowerMock.expectLastCall();
-
         // get the initial assignment
-        EasyMock.expect(member.memberId()).andStubReturn("leader");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+        when(member.memberId()).thenReturn("leader");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
         expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true);
         expectConfigRefreshAndSnapshot(SNAPSHOT);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
+
+        when(statusBackingStore.connectors()).thenReturn(Collections.emptySet());
+        doNothing().when(member).poll(anyLong());
+
+        herder.tick();
 
         // now handle the connector restart
-        member.wakeup();
-        PowerMock.expectLastCall();
-        member.ensureActive();
-        PowerMock.expectLastCall();
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
+        doNothing().when(member).ensureActive();
 
-        PowerMock.replayAll();
+        RestartPlan restartPlan = mock(RestartPlan.class);
+        ConnectorStateInfo connectorStateInfo = mock(ConnectorStateInfo.class);
+        when(restartPlan.restartConnectorStateInfo()).thenReturn(connectorStateInfo);
+        RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
+        doReturn(Optional.of(restartPlan)).when(herder).buildRestartPlan(restartRequest);
+        doNothing().when(configBackingStore).putRestartRequest(restartRequest);
 
-        herder.tick();
         FutureCallback<ConnectorStateInfo> callback = new FutureCallback<>();
         herder.restartConnectorAndTasks(restartRequest, callback);
         herder.tick();
         assertEquals(connectorStateInfo,  callback.get(1000L, TimeUnit.MILLISECONDS));
-        PowerMock.verifyAll();
+
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testDoRestartConnectorAndTasksEmptyPlan() {
         RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
-        EasyMock.expect(herder.buildRestartPlan(restartRequest)).andReturn(Optional.empty()).anyTimes();
-
-        PowerMock.replayAll();
-
+        doReturn(Optional.empty()).when(herder).buildRestartPlan(restartRequest);
         herder.doRestartConnectorAndTasks(restartRequest);
-        PowerMock.verifyAll();
+
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testDoRestartConnectorAndTasksNoAssignments() {
         ConnectorTaskId taskId = new ConnectorTaskId(CONN1, 0);
         RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
-        RestartPlan restartPlan = PowerMock.createMock(RestartPlan.class);
-        EasyMock.expect(restartPlan.shouldRestartConnector()).andReturn(true).anyTimes();
-        EasyMock.expect(restartPlan.shouldRestartTasks()).andReturn(true).anyTimes();
-        EasyMock.expect(restartPlan.taskIdsToRestart()).andReturn(Collections.singletonList(taskId)).anyTimes();
+        RestartPlan restartPlan = mock(RestartPlan.class);
+        when(restartPlan.shouldRestartConnector()).thenReturn(true);
+        when(restartPlan.taskIdsToRestart()).thenReturn(Collections.singletonList(taskId));
 
-        EasyMock.expect(herder.buildRestartPlan(restartRequest)).andReturn(Optional.of(restartPlan)).anyTimes();
+        doReturn(Optional.of(restartPlan)).when(herder).buildRestartPlan(restartRequest);
 
-        PowerMock.replayAll();
         herder.assignment = ExtendedAssignment.empty();
         herder.doRestartConnectorAndTasks(restartRequest);
-        PowerMock.verifyAll();
+
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testDoRestartConnectorAndTasksOnlyConnector() {
         ConnectorTaskId taskId = new ConnectorTaskId(CONN1, 0);
         RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
-        RestartPlan restartPlan = PowerMock.createMock(RestartPlan.class);
-        EasyMock.expect(restartPlan.shouldRestartConnector()).andReturn(true).anyTimes();
-        EasyMock.expect(restartPlan.shouldRestartTasks()).andReturn(true).anyTimes();

Review Comment:
   Good point, done.



##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java:
##########
@@ -1466,560 +1032,421 @@ public void testRestartConnectorAndTasksUnknownConnector() throws Exception {
         assertTrue(ee.getCause() instanceof NotFoundException);
         assertTrue(ee.getMessage().contains("Unknown connector:"));
 
-        PowerMock.verifyAll();
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testRestartConnectorAndTasksNotLeader() throws Exception {
-        RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
-
         // get the initial assignment
-        EasyMock.expect(member.memberId()).andStubReturn("member");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+        when(member.memberId()).thenReturn("member");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
         expectRebalance(1, Collections.emptyList(), Collections.emptyList());
         expectConfigRefreshAndSnapshot(SNAPSHOT);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
+        doNothing().when(member).poll(anyLong());
 
-        // now handle the connector restart
-        member.wakeup();
-        PowerMock.expectLastCall();
-        member.ensureActive();
-        PowerMock.expectLastCall();
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
+        herder.tick();
 
-        PowerMock.replayAll();
+        // now handle the connector restart
+        doNothing().when(member).ensureActive();
 
         herder.tick();
         FutureCallback<ConnectorStateInfo> callback = new FutureCallback<>();
+        RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
         herder.restartConnectorAndTasks(restartRequest, callback);
         herder.tick();
         ExecutionException ee = assertThrows(ExecutionException.class, () -> callback.get(1000L, TimeUnit.MILLISECONDS));
         assertTrue(ee.getCause() instanceof NotLeaderException);
 
-        PowerMock.verifyAll();
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testRestartConnectorAndTasksUnknownStatus() throws Exception {
-        RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
-        EasyMock.expect(herder.buildRestartPlan(restartRequest)).andReturn(Optional.empty()).anyTimes();
-
-        configBackingStore.putRestartRequest(restartRequest);
-        PowerMock.expectLastCall();
-
         // get the initial assignment
-        EasyMock.expect(member.memberId()).andStubReturn("leader");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+        when(member.memberId()).thenReturn("leader");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
         expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true);
         expectConfigRefreshAndSnapshot(SNAPSHOT);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
-
-        // now handle the connector restart
-        member.wakeup();
-        PowerMock.expectLastCall();
-        member.ensureActive();
-        PowerMock.expectLastCall();
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
 
-        PowerMock.replayAll();
+        when(statusBackingStore.connectors()).thenReturn(Collections.emptySet());
+        doNothing().when(member).poll(anyLong());
 
         herder.tick();
+
+        // now handle the connector restart
+        doNothing().when(member).ensureActive();
+        when(statusBackingStore.get(CONN1)).thenReturn(null);
+        RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
+        doNothing().when(configBackingStore).putRestartRequest(restartRequest);
+
         FutureCallback<ConnectorStateInfo> callback = new FutureCallback<>();
         herder.restartConnectorAndTasks(restartRequest, callback);
         herder.tick();
         ExecutionException ee = assertThrows(ExecutionException.class, () -> callback.get(1000L, TimeUnit.MILLISECONDS));
         assertTrue(ee.getCause() instanceof NotFoundException);
         assertTrue(ee.getMessage().contains("Status for connector"));
-        PowerMock.verifyAll();
+
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testRestartConnectorAndTasksSuccess() throws Exception {
-        RestartPlan restartPlan = PowerMock.createMock(RestartPlan.class);
-        ConnectorStateInfo connectorStateInfo = PowerMock.createMock(ConnectorStateInfo.class);
-        EasyMock.expect(restartPlan.restartConnectorStateInfo()).andReturn(connectorStateInfo).anyTimes();
-
-        RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
-        EasyMock.expect(herder.buildRestartPlan(restartRequest)).andReturn(Optional.of(restartPlan)).anyTimes();
-
-        configBackingStore.putRestartRequest(restartRequest);
-        PowerMock.expectLastCall();
-
         // get the initial assignment
-        EasyMock.expect(member.memberId()).andStubReturn("leader");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+        when(member.memberId()).thenReturn("leader");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
         expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true);
         expectConfigRefreshAndSnapshot(SNAPSHOT);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
+
+        when(statusBackingStore.connectors()).thenReturn(Collections.emptySet());
+        doNothing().when(member).poll(anyLong());
+
+        herder.tick();
 
         // now handle the connector restart
-        member.wakeup();
-        PowerMock.expectLastCall();
-        member.ensureActive();
-        PowerMock.expectLastCall();
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
+        doNothing().when(member).ensureActive();
 
-        PowerMock.replayAll();
+        RestartPlan restartPlan = mock(RestartPlan.class);
+        ConnectorStateInfo connectorStateInfo = mock(ConnectorStateInfo.class);
+        when(restartPlan.restartConnectorStateInfo()).thenReturn(connectorStateInfo);
+        RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
+        doReturn(Optional.of(restartPlan)).when(herder).buildRestartPlan(restartRequest);
+        doNothing().when(configBackingStore).putRestartRequest(restartRequest);
 
-        herder.tick();
         FutureCallback<ConnectorStateInfo> callback = new FutureCallback<>();
         herder.restartConnectorAndTasks(restartRequest, callback);
         herder.tick();
         assertEquals(connectorStateInfo,  callback.get(1000L, TimeUnit.MILLISECONDS));
-        PowerMock.verifyAll();
+
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testDoRestartConnectorAndTasksEmptyPlan() {
         RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
-        EasyMock.expect(herder.buildRestartPlan(restartRequest)).andReturn(Optional.empty()).anyTimes();
-
-        PowerMock.replayAll();
-
+        doReturn(Optional.empty()).when(herder).buildRestartPlan(restartRequest);
         herder.doRestartConnectorAndTasks(restartRequest);
-        PowerMock.verifyAll();
+
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testDoRestartConnectorAndTasksNoAssignments() {
         ConnectorTaskId taskId = new ConnectorTaskId(CONN1, 0);
         RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
-        RestartPlan restartPlan = PowerMock.createMock(RestartPlan.class);
-        EasyMock.expect(restartPlan.shouldRestartConnector()).andReturn(true).anyTimes();
-        EasyMock.expect(restartPlan.shouldRestartTasks()).andReturn(true).anyTimes();
-        EasyMock.expect(restartPlan.taskIdsToRestart()).andReturn(Collections.singletonList(taskId)).anyTimes();
+        RestartPlan restartPlan = mock(RestartPlan.class);
+        when(restartPlan.shouldRestartConnector()).thenReturn(true);
+        when(restartPlan.taskIdsToRestart()).thenReturn(Collections.singletonList(taskId));
 
-        EasyMock.expect(herder.buildRestartPlan(restartRequest)).andReturn(Optional.of(restartPlan)).anyTimes();
+        doReturn(Optional.of(restartPlan)).when(herder).buildRestartPlan(restartRequest);
 
-        PowerMock.replayAll();
         herder.assignment = ExtendedAssignment.empty();
         herder.doRestartConnectorAndTasks(restartRequest);
-        PowerMock.verifyAll();
+
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testDoRestartConnectorAndTasksOnlyConnector() {
         ConnectorTaskId taskId = new ConnectorTaskId(CONN1, 0);
         RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
-        RestartPlan restartPlan = PowerMock.createMock(RestartPlan.class);
-        EasyMock.expect(restartPlan.shouldRestartConnector()).andReturn(true).anyTimes();
-        EasyMock.expect(restartPlan.shouldRestartTasks()).andReturn(true).anyTimes();
-        EasyMock.expect(restartPlan.taskIdsToRestart()).andReturn(Collections.singletonList(taskId)).anyTimes();
+        RestartPlan restartPlan = mock(RestartPlan.class);
+        when(restartPlan.shouldRestartConnector()).thenReturn(true);
+        when(restartPlan.taskIdsToRestart()).thenReturn(Collections.singletonList(taskId));
 
-        EasyMock.expect(herder.buildRestartPlan(restartRequest)).andReturn(Optional.of(restartPlan)).anyTimes();
+        doReturn(Optional.of(restartPlan)).when(herder).buildRestartPlan(restartRequest);
 
-        herder.assignment = PowerMock.createMock(ExtendedAssignment.class);
-        EasyMock.expect(herder.assignment.connectors()).andReturn(Collections.singletonList(CONN1)).anyTimes();
-        EasyMock.expect(herder.assignment.tasks()).andReturn(Collections.emptyList()).anyTimes();
+        herder.assignment = mock(ExtendedAssignment.class);
+        when(herder.assignment.connectors()).thenReturn(Collections.singletonList(CONN1));
+        when(herder.assignment.tasks()).thenReturn(Collections.emptyList());
 
-        worker.stopAndAwaitConnector(CONN1);
-        PowerMock.expectLastCall();
+        doNothing().when(worker).stopAndAwaitConnector(CONN1);
 
-        Capture<Callback<TargetState>>  stateCallback = newCapture();
-        worker.startConnector(EasyMock.eq(CONN1), EasyMock.anyObject(), EasyMock.anyObject(),
-                EasyMock.eq(herder), EasyMock.anyObject(TargetState.class), capture(stateCallback));
-        PowerMock.expectLastCall().andAnswer(() -> {
+        ArgumentCaptor<Callback<TargetState>>  stateCallback = ArgumentCaptor.forClass(Callback.class);
+        doAnswer(invocation -> {
             stateCallback.getValue().onCompletion(null, TargetState.STARTED);
             return true;
-        });
-        member.wakeup();
-        PowerMock.expectLastCall();
+        }).when(worker).startConnector(eq(CONN1), any(), any(), eq(herder), any(), stateCallback.capture());
+        doNothing().when(member).wakeup();
+        doNothing().when(statusBackingStore).put(any(ConnectorStatus.class));

Review Comment:
   > One could be to set up a mock expectation with the exact status we expect to see reported to the status store
   
   Makes sense, I've made this change here and in the subsequent tests as well. I guess I just put in the lowest impedance change initially since it's additional coverage (but was required to satisfy the new stricter verifications) 😅 



##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java:
##########
@@ -1466,560 +1032,421 @@ public void testRestartConnectorAndTasksUnknownConnector() throws Exception {
         assertTrue(ee.getCause() instanceof NotFoundException);
         assertTrue(ee.getMessage().contains("Unknown connector:"));
 
-        PowerMock.verifyAll();
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testRestartConnectorAndTasksNotLeader() throws Exception {
-        RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
-
         // get the initial assignment
-        EasyMock.expect(member.memberId()).andStubReturn("member");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+        when(member.memberId()).thenReturn("member");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
         expectRebalance(1, Collections.emptyList(), Collections.emptyList());
         expectConfigRefreshAndSnapshot(SNAPSHOT);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
+        doNothing().when(member).poll(anyLong());
 
-        // now handle the connector restart
-        member.wakeup();
-        PowerMock.expectLastCall();
-        member.ensureActive();
-        PowerMock.expectLastCall();
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
+        herder.tick();
 
-        PowerMock.replayAll();
+        // now handle the connector restart
+        doNothing().when(member).ensureActive();
 
         herder.tick();
         FutureCallback<ConnectorStateInfo> callback = new FutureCallback<>();
+        RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
         herder.restartConnectorAndTasks(restartRequest, callback);
         herder.tick();
         ExecutionException ee = assertThrows(ExecutionException.class, () -> callback.get(1000L, TimeUnit.MILLISECONDS));
         assertTrue(ee.getCause() instanceof NotLeaderException);
 
-        PowerMock.verifyAll();
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testRestartConnectorAndTasksUnknownStatus() throws Exception {
-        RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
-        EasyMock.expect(herder.buildRestartPlan(restartRequest)).andReturn(Optional.empty()).anyTimes();
-
-        configBackingStore.putRestartRequest(restartRequest);
-        PowerMock.expectLastCall();
-
         // get the initial assignment
-        EasyMock.expect(member.memberId()).andStubReturn("leader");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+        when(member.memberId()).thenReturn("leader");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
         expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true);
         expectConfigRefreshAndSnapshot(SNAPSHOT);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
-
-        // now handle the connector restart
-        member.wakeup();
-        PowerMock.expectLastCall();
-        member.ensureActive();
-        PowerMock.expectLastCall();
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
 
-        PowerMock.replayAll();
+        when(statusBackingStore.connectors()).thenReturn(Collections.emptySet());
+        doNothing().when(member).poll(anyLong());
 
         herder.tick();
+
+        // now handle the connector restart
+        doNothing().when(member).ensureActive();
+        when(statusBackingStore.get(CONN1)).thenReturn(null);
+        RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
+        doNothing().when(configBackingStore).putRestartRequest(restartRequest);
+
         FutureCallback<ConnectorStateInfo> callback = new FutureCallback<>();
         herder.restartConnectorAndTasks(restartRequest, callback);
         herder.tick();
         ExecutionException ee = assertThrows(ExecutionException.class, () -> callback.get(1000L, TimeUnit.MILLISECONDS));
         assertTrue(ee.getCause() instanceof NotFoundException);
         assertTrue(ee.getMessage().contains("Status for connector"));
-        PowerMock.verifyAll();
+
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testRestartConnectorAndTasksSuccess() throws Exception {
-        RestartPlan restartPlan = PowerMock.createMock(RestartPlan.class);
-        ConnectorStateInfo connectorStateInfo = PowerMock.createMock(ConnectorStateInfo.class);
-        EasyMock.expect(restartPlan.restartConnectorStateInfo()).andReturn(connectorStateInfo).anyTimes();
-
-        RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
-        EasyMock.expect(herder.buildRestartPlan(restartRequest)).andReturn(Optional.of(restartPlan)).anyTimes();
-
-        configBackingStore.putRestartRequest(restartRequest);
-        PowerMock.expectLastCall();
-
         // get the initial assignment
-        EasyMock.expect(member.memberId()).andStubReturn("leader");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+        when(member.memberId()).thenReturn("leader");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
         expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true);
         expectConfigRefreshAndSnapshot(SNAPSHOT);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
+
+        when(statusBackingStore.connectors()).thenReturn(Collections.emptySet());
+        doNothing().when(member).poll(anyLong());
+
+        herder.tick();
 
         // now handle the connector restart
-        member.wakeup();
-        PowerMock.expectLastCall();
-        member.ensureActive();
-        PowerMock.expectLastCall();
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
+        doNothing().when(member).ensureActive();
 
-        PowerMock.replayAll();
+        RestartPlan restartPlan = mock(RestartPlan.class);
+        ConnectorStateInfo connectorStateInfo = mock(ConnectorStateInfo.class);
+        when(restartPlan.restartConnectorStateInfo()).thenReturn(connectorStateInfo);
+        RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
+        doReturn(Optional.of(restartPlan)).when(herder).buildRestartPlan(restartRequest);
+        doNothing().when(configBackingStore).putRestartRequest(restartRequest);
 
-        herder.tick();
         FutureCallback<ConnectorStateInfo> callback = new FutureCallback<>();
         herder.restartConnectorAndTasks(restartRequest, callback);
         herder.tick();
         assertEquals(connectorStateInfo,  callback.get(1000L, TimeUnit.MILLISECONDS));
-        PowerMock.verifyAll();
+
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testDoRestartConnectorAndTasksEmptyPlan() {
         RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
-        EasyMock.expect(herder.buildRestartPlan(restartRequest)).andReturn(Optional.empty()).anyTimes();
-
-        PowerMock.replayAll();
-
+        doReturn(Optional.empty()).when(herder).buildRestartPlan(restartRequest);
         herder.doRestartConnectorAndTasks(restartRequest);
-        PowerMock.verifyAll();
+
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testDoRestartConnectorAndTasksNoAssignments() {
         ConnectorTaskId taskId = new ConnectorTaskId(CONN1, 0);
         RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
-        RestartPlan restartPlan = PowerMock.createMock(RestartPlan.class);
-        EasyMock.expect(restartPlan.shouldRestartConnector()).andReturn(true).anyTimes();
-        EasyMock.expect(restartPlan.shouldRestartTasks()).andReturn(true).anyTimes();
-        EasyMock.expect(restartPlan.taskIdsToRestart()).andReturn(Collections.singletonList(taskId)).anyTimes();
+        RestartPlan restartPlan = mock(RestartPlan.class);
+        when(restartPlan.shouldRestartConnector()).thenReturn(true);
+        when(restartPlan.taskIdsToRestart()).thenReturn(Collections.singletonList(taskId));
 
-        EasyMock.expect(herder.buildRestartPlan(restartRequest)).andReturn(Optional.of(restartPlan)).anyTimes();
+        doReturn(Optional.of(restartPlan)).when(herder).buildRestartPlan(restartRequest);
 
-        PowerMock.replayAll();
         herder.assignment = ExtendedAssignment.empty();
         herder.doRestartConnectorAndTasks(restartRequest);
-        PowerMock.verifyAll();
+
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testDoRestartConnectorAndTasksOnlyConnector() {
         ConnectorTaskId taskId = new ConnectorTaskId(CONN1, 0);
         RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
-        RestartPlan restartPlan = PowerMock.createMock(RestartPlan.class);
-        EasyMock.expect(restartPlan.shouldRestartConnector()).andReturn(true).anyTimes();
-        EasyMock.expect(restartPlan.shouldRestartTasks()).andReturn(true).anyTimes();
-        EasyMock.expect(restartPlan.taskIdsToRestart()).andReturn(Collections.singletonList(taskId)).anyTimes();
+        RestartPlan restartPlan = mock(RestartPlan.class);
+        when(restartPlan.shouldRestartConnector()).thenReturn(true);
+        when(restartPlan.taskIdsToRestart()).thenReturn(Collections.singletonList(taskId));
 
-        EasyMock.expect(herder.buildRestartPlan(restartRequest)).andReturn(Optional.of(restartPlan)).anyTimes();
+        doReturn(Optional.of(restartPlan)).when(herder).buildRestartPlan(restartRequest);
 
-        herder.assignment = PowerMock.createMock(ExtendedAssignment.class);
-        EasyMock.expect(herder.assignment.connectors()).andReturn(Collections.singletonList(CONN1)).anyTimes();
-        EasyMock.expect(herder.assignment.tasks()).andReturn(Collections.emptyList()).anyTimes();
+        herder.assignment = mock(ExtendedAssignment.class);
+        when(herder.assignment.connectors()).thenReturn(Collections.singletonList(CONN1));
+        when(herder.assignment.tasks()).thenReturn(Collections.emptyList());
 
-        worker.stopAndAwaitConnector(CONN1);
-        PowerMock.expectLastCall();
+        doNothing().when(worker).stopAndAwaitConnector(CONN1);
 
-        Capture<Callback<TargetState>>  stateCallback = newCapture();
-        worker.startConnector(EasyMock.eq(CONN1), EasyMock.anyObject(), EasyMock.anyObject(),
-                EasyMock.eq(herder), EasyMock.anyObject(TargetState.class), capture(stateCallback));
-        PowerMock.expectLastCall().andAnswer(() -> {
+        ArgumentCaptor<Callback<TargetState>>  stateCallback = ArgumentCaptor.forClass(Callback.class);
+        doAnswer(invocation -> {
             stateCallback.getValue().onCompletion(null, TargetState.STARTED);
             return true;
-        });
-        member.wakeup();
-        PowerMock.expectLastCall();
+        }).when(worker).startConnector(eq(CONN1), any(), any(), eq(herder), any(), stateCallback.capture());
+        doNothing().when(member).wakeup();
+        doNothing().when(statusBackingStore).put(any(ConnectorStatus.class));
 
-
-        herder.onRestart(CONN1);
-        EasyMock.expectLastCall();
-
-        PowerMock.replayAll();
         herder.doRestartConnectorAndTasks(restartRequest);
-        PowerMock.verifyAll();
+
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testDoRestartConnectorAndTasksOnlyTasks() {
         RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
-        RestartPlan restartPlan = PowerMock.createMock(RestartPlan.class);
-        EasyMock.expect(restartPlan.shouldRestartConnector()).andReturn(true).anyTimes();
-        EasyMock.expect(restartPlan.shouldRestartTasks()).andReturn(true).anyTimes();
+        RestartPlan restartPlan = mock(RestartPlan.class);
+        when(restartPlan.shouldRestartConnector()).thenReturn(true);
         // The connector has three tasks
-        EasyMock.expect(restartPlan.taskIdsToRestart()).andReturn(Arrays.asList(TASK0, TASK1, TASK2)).anyTimes();
-        EasyMock.expect(restartPlan.restartTaskCount()).andReturn(3).anyTimes();
-        EasyMock.expect(restartPlan.totalTaskCount()).andReturn(3).anyTimes();
-        EasyMock.expect(herder.buildRestartPlan(restartRequest)).andReturn(Optional.of(restartPlan)).anyTimes();
+        when(restartPlan.taskIdsToRestart()).thenReturn(Arrays.asList(TASK0, TASK1, TASK2));
+        when(restartPlan.totalTaskCount()).thenReturn(3);
+        doReturn(Optional.of(restartPlan)).when(herder).buildRestartPlan(restartRequest);
 
-        herder.assignment = PowerMock.createMock(ExtendedAssignment.class);
-        EasyMock.expect(herder.assignment.connectors()).andReturn(Collections.emptyList()).anyTimes();
+        herder.assignment = mock(ExtendedAssignment.class);
+        when(herder.assignment.connectors()).thenReturn(Collections.emptyList());
         // But only one task is assigned to this worker
-        EasyMock.expect(herder.assignment.tasks()).andReturn(Collections.singletonList(TASK0)).anyTimes();
+        when(herder.assignment.tasks()).thenReturn(Collections.singletonList(TASK0));
 
         herder.configState = SNAPSHOT;
 
-        worker.stopAndAwaitTasks(Collections.singletonList(TASK0));
-        PowerMock.expectLastCall();
-
-        herder.onRestart(TASK0);
-        EasyMock.expectLastCall();
-
-        worker.startSourceTask(EasyMock.eq(TASK0), EasyMock.anyObject(), EasyMock.anyObject(), EasyMock.anyObject(),
-                EasyMock.eq(herder), EasyMock.anyObject(TargetState.class));
-        PowerMock.expectLastCall().andReturn(true);
+        doNothing().when(worker).stopAndAwaitTasks(Collections.singletonList(TASK0));
+        doNothing().when(statusBackingStore).put(any(TaskStatus.class));
+        when(worker.startSourceTask(eq(TASK0), any(), any(), any(), eq(herder), any())).thenReturn(true);
 
-        PowerMock.replayAll();
         herder.doRestartConnectorAndTasks(restartRequest);
-        PowerMock.verifyAll();
+
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testDoRestartConnectorAndTasksBoth() {
         ConnectorTaskId taskId = new ConnectorTaskId(CONN1, 0);
         RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
-        RestartPlan restartPlan = PowerMock.createMock(RestartPlan.class);
-        EasyMock.expect(restartPlan.shouldRestartConnector()).andReturn(true).anyTimes();
-        EasyMock.expect(restartPlan.shouldRestartTasks()).andReturn(true).anyTimes();
-        EasyMock.expect(restartPlan.taskIdsToRestart()).andReturn(Collections.singletonList(taskId)).anyTimes();
-        EasyMock.expect(restartPlan.restartTaskCount()).andReturn(1).anyTimes();
-        EasyMock.expect(restartPlan.totalTaskCount()).andReturn(1).anyTimes();
-        EasyMock.expect(herder.buildRestartPlan(restartRequest)).andReturn(Optional.of(restartPlan)).anyTimes();
-
-        herder.assignment = PowerMock.createMock(ExtendedAssignment.class);
-        EasyMock.expect(herder.assignment.connectors()).andReturn(Collections.singletonList(CONN1)).anyTimes();
-        EasyMock.expect(herder.assignment.tasks()).andReturn(Collections.singletonList(taskId)).anyTimes();
-
-        herder.configState = SNAPSHOT;
+        RestartPlan restartPlan = mock(RestartPlan.class);
+        when(restartPlan.shouldRestartConnector()).thenReturn(true);
+        when(restartPlan.taskIdsToRestart()).thenReturn(Collections.singletonList(taskId));
+        when(restartPlan.totalTaskCount()).thenReturn(1);
+        doReturn(Optional.of(restartPlan)).when(herder).buildRestartPlan(restartRequest);
 
-        worker.stopAndAwaitConnector(CONN1);
-        PowerMock.expectLastCall();
-
-        Capture<Callback<TargetState>>  stateCallback = newCapture();
-        worker.startConnector(EasyMock.eq(CONN1), EasyMock.anyObject(), EasyMock.anyObject(),
-                EasyMock.eq(herder), EasyMock.anyObject(TargetState.class), capture(stateCallback));
+        herder.assignment = mock(ExtendedAssignment.class);
+        when(herder.assignment.connectors()).thenReturn(Collections.singletonList(CONN1));
+        when(herder.assignment.tasks()).thenReturn(Collections.singletonList(taskId));
 
+        herder.configState = SNAPSHOT;
 
-        herder.onRestart(CONN1);
-        EasyMock.expectLastCall();
+        doNothing().when(worker).stopAndAwaitConnector(CONN1);
 
-        worker.stopAndAwaitTasks(Collections.singletonList(taskId));
-        PowerMock.expectLastCall();
+        ArgumentCaptor<Callback<TargetState>>  stateCallback = ArgumentCaptor.forClass(Callback.class);
+        doAnswer(invocation -> {
+            stateCallback.getValue().onCompletion(null, TargetState.STARTED);
+            return true;
+        }).when(worker).startConnector(eq(CONN1), any(), any(), eq(herder), any(), stateCallback.capture());
+        doNothing().when(member).wakeup();
+        doNothing().when(statusBackingStore).put(any(ConnectorStatus.class));
 
-        herder.onRestart(taskId);
-        EasyMock.expectLastCall();
+        doNothing().when(worker).stopAndAwaitTasks(Collections.singletonList(taskId));
 
-        worker.startSourceTask(EasyMock.eq(TASK0), EasyMock.anyObject(), EasyMock.anyObject(), EasyMock.anyObject(),
-                EasyMock.eq(herder), EasyMock.anyObject(TargetState.class));
-        PowerMock.expectLastCall().andReturn(true);
+        when(worker.startSourceTask(eq(TASK0), any(), any(), any(), eq(herder), any())).thenReturn(true);
+        doNothing().when(statusBackingStore).put(any(TaskStatus.class));
 
-        PowerMock.replayAll();
         herder.doRestartConnectorAndTasks(restartRequest);
-        PowerMock.verifyAll();
+
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testRestartTask() throws Exception {
-        EasyMock.expect(worker.connectorTaskConfigs(CONN1, conn1SinkConfig)).andStubReturn(TASK_CONFIGS);
-
         // get the initial assignment
-        EasyMock.expect(member.memberId()).andStubReturn("leader");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+        when(member.memberId()).thenReturn("leader");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
         expectRebalance(1, Collections.emptyList(), singletonList(TASK0), true);
         expectConfigRefreshAndSnapshot(SNAPSHOT);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
-        worker.startSourceTask(EasyMock.eq(TASK0), EasyMock.anyObject(), EasyMock.anyObject(), EasyMock.anyObject(),
-                EasyMock.eq(herder), EasyMock.eq(TargetState.STARTED));
-        PowerMock.expectLastCall().andReturn(true);
-
-        // now handle the task restart
-        member.wakeup();
-        PowerMock.expectLastCall();
-        member.ensureActive();
-        PowerMock.expectLastCall();
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
 
-        worker.stopAndAwaitTask(TASK0);
-        PowerMock.expectLastCall();
-        worker.startSourceTask(EasyMock.eq(TASK0), EasyMock.anyObject(), EasyMock.anyObject(), EasyMock.anyObject(),
-                EasyMock.eq(herder), EasyMock.eq(TargetState.STARTED));
-        PowerMock.expectLastCall().andReturn(true);
+        when(statusBackingStore.connectors()).thenReturn(Collections.emptySet());
+        doNothing().when(member).poll(anyLong());
 
-        PowerMock.replayAll();
+        when(worker.startSourceTask(eq(TASK0), any(), any(), any(), eq(herder), any())).thenReturn(true);
 
         herder.tick();
+
+        // now handle the task restart
+        doNothing().when(member).ensureActive();
+        doNothing().when(worker).stopAndAwaitTask(TASK0);
         FutureCallback<Void> callback = new FutureCallback<>();
         herder.restartTask(TASK0, callback);
         herder.tick();
         callback.get(1000L, TimeUnit.MILLISECONDS);
 
-        PowerMock.verifyAll();
+        verify(worker, times(2)).startSourceTask(eq(TASK0), any(), any(), any(), eq(herder), any());
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testRestartUnknownTask() throws Exception {
         // get the initial assignment
-        EasyMock.expect(member.memberId()).andStubReturn("member");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+        when(member.memberId()).thenReturn("member");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
         expectRebalance(1, Collections.emptyList(), Collections.emptyList());
         expectConfigRefreshAndSnapshot(SNAPSHOT);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
-
-        member.wakeup();
-        PowerMock.expectLastCall();
-        member.ensureActive();
-        PowerMock.expectLastCall();
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
+        doNothing().when(member).poll(anyLong());
 
-        PowerMock.replayAll();
+        herder.tick();
 
+        doNothing().when(member).ensureActive();
         FutureCallback<Void> callback = new FutureCallback<>();
-        herder.tick();
         herder.restartTask(new ConnectorTaskId("blah", 0), callback);
         herder.tick();
 
-        try {
-            callback.get(1000L, TimeUnit.MILLISECONDS);
-            fail("Expected NotLeaderException to be raised");
-        } catch (ExecutionException e) {
-            assertTrue(e.getCause() instanceof NotFoundException);
-        }
+        ExecutionException e = assertThrows(ExecutionException.class, () -> callback.get(1000, TimeUnit.MILLISECONDS));
+        assertEquals(NotFoundException.class, e.getCause().getClass());
 
-        PowerMock.verifyAll();
-    }
-
-    @Test
-    public void testRequestProcessingOrder() {
-        final DistributedHerder.DistributedHerderRequest req1 = herder.addRequest(100, null, null);
-        final DistributedHerder.DistributedHerderRequest req2 = herder.addRequest(10, null, null);
-        final DistributedHerder.DistributedHerderRequest req3 = herder.addRequest(200, null, null);
-        final DistributedHerder.DistributedHerderRequest req4 = herder.addRequest(200, null, null);
-
-        assertEquals(req2, herder.requests.pollFirst()); // lowest delay
-        assertEquals(req1, herder.requests.pollFirst()); // next lowest delay
-        assertEquals(req3, herder.requests.pollFirst()); // same delay as req4, but added first
-        assertEquals(req4, herder.requests.pollFirst());
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testRestartTaskRedirectToLeader() throws Exception {
         // get the initial assignment
-        EasyMock.expect(member.memberId()).andStubReturn("member");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+        when(member.memberId()).thenReturn("member");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
         expectRebalance(1, Collections.emptyList(), Collections.emptyList());
         expectConfigRefreshAndSnapshot(SNAPSHOT);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
-
-        // now handle the task restart
-        member.wakeup();
-        PowerMock.expectLastCall();
-        member.ensureActive();
-        PowerMock.expectLastCall();
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
-
-        PowerMock.replayAll();
+        doNothing().when(member).poll(anyLong());
 
         herder.tick();
+
+        // now handle the task restart
+        doNothing().when(member).ensureActive();
         FutureCallback<Void> callback = new FutureCallback<>();
         herder.restartTask(TASK0, callback);
         herder.tick();
 
-        try {
-            callback.get(1000L, TimeUnit.MILLISECONDS);
-            fail("Expected NotLeaderException to be raised");
-        } catch (ExecutionException e) {
-            assertTrue(e.getCause() instanceof NotLeaderException);
-        }
+        ExecutionException e = assertThrows(ExecutionException.class, () -> callback.get(1000, TimeUnit.MILLISECONDS));
+        assertEquals(NotLeaderException.class, e.getCause().getClass());
 
-        PowerMock.verifyAll();
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testRestartTaskRedirectToOwner() throws Exception {
         // get the initial assignment
-        EasyMock.expect(member.memberId()).andStubReturn("leader");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+        when(member.memberId()).thenReturn("leader");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
         expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true);
         expectConfigRefreshAndSnapshot(SNAPSHOT);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
+
+        when(statusBackingStore.connectors()).thenReturn(Collections.emptySet());
+        doNothing().when(member).poll(anyLong());
+
+        herder.tick();
 
         // now handle the task restart
         String ownerUrl = "ownerUrl";
-        EasyMock.expect(member.ownerUrl(TASK0)).andReturn(ownerUrl);
-        member.wakeup();
-        PowerMock.expectLastCall();
-        member.ensureActive();
-        PowerMock.expectLastCall();
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
+        when(member.ownerUrl(TASK0)).thenReturn(ownerUrl);
+        doNothing().when(member).ensureActive();
 
-        PowerMock.replayAll();
-
-        herder.tick();
         FutureCallback<Void> callback = new FutureCallback<>();
         herder.restartTask(TASK0, callback);
         herder.tick();
 
-        try {
-            callback.get(1000L, TimeUnit.MILLISECONDS);
-            fail("Expected NotLeaderException to be raised");
-        } catch (ExecutionException e) {
-            assertTrue(e.getCause() instanceof NotAssignedException);
-            NotAssignedException notAssignedException = (NotAssignedException) e.getCause();
-            assertEquals(ownerUrl, notAssignedException.forwardUrl());
-        }
+        ExecutionException e = assertThrows(ExecutionException.class, () -> callback.get(1000, TimeUnit.MILLISECONDS));
+        assertEquals(NotAssignedException.class, e.getCause().getClass());
+        assertEquals(ownerUrl, ((NotAssignedException) e.getCause()).forwardUrl());
 
-        PowerMock.verifyAll();
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
-    public void testConnectorConfigAdded() {
+    public void testRequestProcessingOrder() {
+        final DistributedHerder.DistributedHerderRequest req1 = herder.addRequest(100, null, null);
+        final DistributedHerder.DistributedHerderRequest req2 = herder.addRequest(10, null, null);
+        final DistributedHerder.DistributedHerderRequest req3 = herder.addRequest(200, null, null);
+        final DistributedHerder.DistributedHerderRequest req4 = herder.addRequest(200, null, null);
+
+        assertEquals(req2, herder.requests.pollFirst()); // lowest delay
+        assertEquals(req1, herder.requests.pollFirst()); // next lowest delay
+        assertEquals(req3, herder.requests.pollFirst()); // same delay as req4, but added first
+        assertEquals(req4, herder.requests.pollFirst());
+    }
+
+    @Test
+    public void testConnectorConfigAdded() throws Exception {
         // If a connector was added, we need to rebalance
-        EasyMock.expect(member.memberId()).andStubReturn("member");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+        when(member.memberId()).thenReturn("member");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
 
         // join, no configs so no need to catch up on config topic
         expectRebalance(-1, Collections.emptyList(), Collections.emptyList());
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
+        doNothing().when(member).poll(anyLong());
+
+        herder.tick(); // join
 
-        // apply config
-        member.wakeup();
-        PowerMock.expectLastCall();
-        member.ensureActive();
-        PowerMock.expectLastCall();
         // Checks for config updates and starts rebalance
-        EasyMock.expect(configBackingStore.snapshot()).andReturn(SNAPSHOT);
-        member.requestRejoin();
-        PowerMock.expectLastCall();
+        doNothing().when(configBackingStore).refresh(anyLong(), any());
+        when(configBackingStore.snapshot()).thenReturn(SNAPSHOT);

Review Comment:
   Whoops 🤦 



##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java:
##########
@@ -3929,543 +3008,636 @@ public void testPollDurationOnSlowConnectorOperations() {
         final int rebalanceDelayMs = 20000;
         final long operationDelayMs = 10000;
         final long maxPollWaitMs = rebalanceDelayMs - operationDelayMs;
-        EasyMock.expect(member.memberId()).andStubReturn("member");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(connectProtocolVersion);
+        when(member.memberId()).thenReturn("member");
+        when(member.currentProtocolVersion()).thenReturn(connectProtocolVersion);
 
         // Assign the connector to this worker, and have it start
         expectRebalance(Collections.emptyList(), Collections.emptyList(), ConnectProtocol.Assignment.NO_ERROR, 1, singletonList(CONN1), Collections.emptyList(), rebalanceDelayMs);
         expectConfigRefreshAndSnapshot(SNAPSHOT);
-        Capture<Callback<TargetState>> onFirstStart = newCapture();
-        worker.startConnector(EasyMock.eq(CONN1), EasyMock.anyObject(), EasyMock.anyObject(),
-                EasyMock.eq(herder), EasyMock.eq(TargetState.STARTED), capture(onFirstStart));
-        PowerMock.expectLastCall().andAnswer(() -> {
+        ArgumentCaptor<Callback<TargetState>> onFirstStart = ArgumentCaptor.forClass(Callback.class);
+        doAnswer(invocation -> {
             time.sleep(operationDelayMs);
             onFirstStart.getValue().onCompletion(null, TargetState.STARTED);
             return true;
-        });
-        member.wakeup();
-        PowerMock.expectLastCall();
-        expectExecuteTaskReconfiguration(true, conn1SinkConfig, () -> TASK_CONFIGS);
-        // We should poll for less than the delay - time to start the connector, meaning that a long connector start
-        // does not delay the poll timeout
-        member.poll(leq(maxPollWaitMs));
-        PowerMock.expectLastCall();
+        }).when(worker).startConnector(eq(CONN1), eq(CONN1_CONFIG), any(), eq(herder), eq(TargetState.STARTED), onFirstStart.capture());
+        expectExecuteTaskReconfiguration(true, conn1SinkConfig, invocation -> TASK_CONFIGS);
+
+        herder.tick();
 
         // Rebalance again due to config update
-        member.wakeup();
-        PowerMock.expectLastCall();
         expectRebalance(Collections.emptyList(), Collections.emptyList(), ConnectProtocol.Assignment.NO_ERROR, 1, singletonList(CONN1), Collections.emptyList(), rebalanceDelayMs);
-        EasyMock.expect(configBackingStore.snapshot()).andReturn(SNAPSHOT_UPDATED_CONN1_CONFIG);
-
-        worker.stopAndAwaitConnector(CONN1);
-        PowerMock.expectLastCall();
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(connectProtocolVersion);
-        Capture<Callback<TargetState>> onSecondStart = newCapture();
-        worker.startConnector(EasyMock.eq(CONN1), EasyMock.anyObject(), EasyMock.anyObject(),
-                EasyMock.eq(herder), EasyMock.eq(TargetState.STARTED), capture(onSecondStart));
-        PowerMock.expectLastCall().andAnswer(() -> {
+        when(configBackingStore.snapshot()).thenReturn(SNAPSHOT_UPDATED_CONN1_CONFIG);
+        doNothing().when(worker).stopAndAwaitConnector(CONN1);
+
+        ArgumentCaptor<Callback<TargetState>> onSecondStart = ArgumentCaptor.forClass(Callback.class);
+        doAnswer(invocation -> {
             time.sleep(operationDelayMs);
             onSecondStart.getValue().onCompletion(null, TargetState.STARTED);
             return true;
-        });
-        member.wakeup();
-        PowerMock.expectLastCall();
-        member.poll(leq(maxPollWaitMs));
-        PowerMock.expectLastCall();
+        }).when(worker).startConnector(eq(CONN1), eq(CONN1_CONFIG_UPDATED), any(), eq(herder), eq(TargetState.STARTED), onSecondStart.capture());
+        expectExecuteTaskReconfiguration(true, conn1SinkConfigUpdated, invocation -> TASK_CONFIGS);
+
+        configUpdateListener.onConnectorConfigUpdate(CONN1); // read updated config
+        herder.tick();
 
         // Third tick should resolve all outstanding requests
         expectRebalance(Collections.emptyList(), Collections.emptyList(), ConnectProtocol.Assignment.NO_ERROR, 1, singletonList(CONN1), Collections.emptyList(), rebalanceDelayMs);
         // which includes querying the connector task configs after the update
-        expectExecuteTaskReconfiguration(true, conn1SinkConfigUpdated, () -> {
+        expectExecuteTaskReconfiguration(true, conn1SinkConfigUpdated, invocation -> {
             time.sleep(operationDelayMs);
             return TASK_CONFIGS;
         });
-        member.poll(leq(maxPollWaitMs));
-        PowerMock.expectLastCall();
-
-        PowerMock.replayAll();
-        herder.tick();
-        configUpdateListener.onConnectorConfigUpdate(CONN1); // read updated config
-        herder.tick();
         herder.tick();
-        PowerMock.verifyAll();
+
+        // We should poll for less than the delay - time to start the connector, meaning that a long connector start
+        // does not delay the poll timeout
+        verify(member, times(3)).poll(leq(maxPollWaitMs));
+        verify(worker, times(2)).startConnector(eq(CONN1), any(), any(), eq(herder), eq(TargetState.STARTED), any());
+        verifyNoMoreInteractions(member, worker, configBackingStore);
     }
 
     @Test
     public void shouldThrowWhenStartAndStopExecutorThrowsRejectedExecutionExceptionAndHerderNotStopping() {
-        EasyMock.expect(member.memberId()).andStubReturn("leader");
+        when(member.memberId()).thenReturn("leader");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
         expectRebalance(1, Arrays.asList(CONN1), Collections.emptyList(), true);
         expectConfigRefreshAndSnapshot(SNAPSHOT);
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
-
-        PowerMock.replayAll();
 
         herder.startAndStopExecutor.shutdown();
         assertThrows(RejectedExecutionException.class, herder::tick);
+    }
+
+    @Test
+    public void testTaskReconfigurationRetriesWithConnectorTaskConfigsException() {
+        when(member.memberId()).thenReturn("leader");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
+        expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true);
+        expectConfigRefreshAndSnapshot(SNAPSHOT);
+
+        when(worker.isRunning(CONN1)).thenReturn(true);
+        when(worker.getPlugins()).thenReturn(plugins);
+
+        SinkConnectorConfig sinkConnectorConfig = new SinkConnectorConfig(plugins, CONN1_CONFIG);
+
+        when(worker.connectorTaskConfigs(CONN1, sinkConnectorConfig))
+                .thenThrow(new ConnectException("Failed to generate task configs"))
+                .thenThrow(new ConnectException("Failed to generate task configs"))
+                .thenReturn(TASK_CONFIGS);
+
+        expectAndVerifyTaskReconfigurationRetries();
+    }
+
+    @Test
+    public void testTaskReconfigurationRetriesWithLeaderRequestForwardingException() {
+        herder = mock(DistributedHerder.class, withSettings().defaultAnswer(CALLS_REAL_METHODS).useConstructor(new DistributedConfig(HERDER_CONFIG),
+                worker, WORKER_ID, KAFKA_CLUSTER_ID, statusBackingStore, configBackingStore, member, MEMBER_URL, restClient, metrics, time,
+                noneConnectorClientConfigOverridePolicy, Collections.emptyList(), new MockSynchronousExecutor(), new AutoCloseable[]{}));
+
+        rebalanceListener = herder.new RebalanceListener(time);
+
+        when(member.memberId()).thenReturn("member");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
+        expectRebalance(1, Collections.emptyList(), Collections.emptyList(), false);
+        expectConfigRefreshAndSnapshot(SNAPSHOT);
+
+        when(worker.isRunning(CONN1)).thenReturn(true);
+        when(worker.getPlugins()).thenReturn(plugins);
+
+        SinkConnectorConfig sinkConnectorConfig = new SinkConnectorConfig(plugins, CONN1_CONFIG);
+
+        List<Map<String, String>> changedTaskConfigs = new ArrayList<>(TASK_CONFIGS);
+        changedTaskConfigs.add(TASK_CONFIG);
+        when(worker.connectorTaskConfigs(CONN1, sinkConnectorConfig)).thenReturn(changedTaskConfigs);
+
+        when(restClient.httpRequest(any(), eq("POST"), any(), any(), any(), any(), any()))
+                .thenThrow(new ConnectException("Request to leader to reconfigure connector tasks failed"))
+                .thenThrow(new ConnectException("Request to leader to reconfigure connector tasks failed"))
+                .thenReturn(null);
+
+        expectAndVerifyTaskReconfigurationRetries();
+    }
+
+    private void expectAndVerifyTaskReconfigurationRetries() {
+        // initial tick
+        herder.tick();
+        herder.requestTaskReconfiguration(CONN1);
+        // process the task reconfiguration request in this tick
+        herder.tick();
+        // advance the time by 250ms so that the task reconfiguration request with initial retry backoff is processed
+        time.sleep(250);
+        herder.tick();
+        // advance the time by 500ms so that the task reconfiguration request with double the initial retry backoff is processed
+        time.sleep(500);
+        herder.tick();
+
+        // 1. end of initial tick when no request has been added to the herder queue yet
+        // 2. the third task reconfiguration request is expected to pass; so expect no more retries (a Long.MAX_VALUE poll
+        //    timeout indicates that there is no herder request currently in the queue)
+        verify(member, times(2)).poll(eq(Long.MAX_VALUE));
+
+        // task reconfiguration herder request with initial retry backoff
+        verify(member).poll(eq(250L));
+
+        // task reconfiguration herder request with double the initial retry backoff
+        verify(member).poll(eq(500L));
+
+        verifyNoMoreInteractions(member, worker);
+    }
+
+    @Test
+    public void processRestartRequestsFailureSuppression() {
+        doNothing().when(member).wakeup();
+
+        final String connectorName = "foo";
+        RestartRequest restartRequest = new RestartRequest(connectorName, false, false);
+        doThrow(new RuntimeException()).when(herder).buildRestartPlan(restartRequest);
+
+        configUpdateListener.onRestartRequest(restartRequest);
+        assertEquals(1, herder.pendingRestartRequests.size());
+        herder.processRestartRequests();
+        assertTrue(herder.pendingRestartRequests.isEmpty());
+
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
+    }
+
+    @Test
+    public void processRestartRequestsDequeue() {
+        doNothing().when(member).wakeup();
+        doReturn(Optional.empty()).when(herder).buildRestartPlan(any(RestartRequest.class));
+
+        RestartRequest restartRequest = new RestartRequest("foo", false, false);
+        configUpdateListener.onRestartRequest(restartRequest);
+        restartRequest = new RestartRequest("bar", false, false);
+        configUpdateListener.onRestartRequest(restartRequest);
+        assertEquals(2, herder.pendingRestartRequests.size());
+        herder.processRestartRequests();
+        assertTrue(herder.pendingRestartRequests.isEmpty());
+    }
+
+    @Test
+    public void preserveHighestImpactRestartRequest() {
+        doNothing().when(member).wakeup();
+
+        final String connectorName = "foo";
+        RestartRequest restartRequest = new RestartRequest(connectorName, false, false);
+        configUpdateListener.onRestartRequest(restartRequest);
+
+        // Will overwrite as this is higher impact
+        restartRequest = new RestartRequest(connectorName, false, true);
+        configUpdateListener.onRestartRequest(restartRequest);
+        assertEquals(1, herder.pendingRestartRequests.size());
+        assertFalse(herder.pendingRestartRequests.get(connectorName).onlyFailed());
+        assertTrue(herder.pendingRestartRequests.get(connectorName).includeTasks());
+
+        // Will be ignored as the existing request has higher impact
+        restartRequest = new RestartRequest(connectorName, true, false);
+        configUpdateListener.onRestartRequest(restartRequest);
+        assertEquals(1, herder.pendingRestartRequests.size());
+        // Compare against existing request
+        assertFalse(herder.pendingRestartRequests.get(connectorName).onlyFailed());
+        assertTrue(herder.pendingRestartRequests.get(connectorName).includeTasks());
+
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
+    }
+
+    @Test
+    public void testExactlyOnceSourceSupportValidation() {
+        herder = exactlyOnceHerder();
+        Map<String, String> config = new HashMap<>();
+        config.put(SourceConnectorConfig.EXACTLY_ONCE_SUPPORT_CONFIG, REQUIRED.toString());
+
+        SourceConnector connectorMock = mock(SourceConnector.class);
+        when(connectorMock.exactlyOnceSupport(eq(config))).thenReturn(ExactlyOnceSupport.SUPPORTED);
+
+        Map<String, ConfigValue> validatedConfigs = herder.validateSourceConnectorConfig(
+                connectorMock, SourceConnectorConfig.configDef(), config);
+
+        List<String> errors = validatedConfigs.get(SourceConnectorConfig.EXACTLY_ONCE_SUPPORT_CONFIG).errorMessages();
+        assertEquals(Collections.emptyList(), errors);
+    }
+
+    @Test
+    public void testExactlyOnceSourceSupportValidationOnUnsupportedConnector() {
+        herder = exactlyOnceHerder();
+        Map<String, String> config = new HashMap<>();
+        config.put(SourceConnectorConfig.EXACTLY_ONCE_SUPPORT_CONFIG, REQUIRED.toString());
+
+        SourceConnector connectorMock = mock(SourceConnector.class);
+        when(connectorMock.exactlyOnceSupport(eq(config))).thenReturn(ExactlyOnceSupport.UNSUPPORTED);
+
+        Map<String, ConfigValue> validatedConfigs = herder.validateSourceConnectorConfig(
+                connectorMock, SourceConnectorConfig.configDef(), config);
+
+        List<String> errors = validatedConfigs.get(SourceConnectorConfig.EXACTLY_ONCE_SUPPORT_CONFIG).errorMessages();
+        assertEquals(
+                Collections.singletonList("The connector does not support exactly-once semantics with the provided configuration."),
+                errors);
+    }
+
+    @Test
+    public void testExactlyOnceSourceSupportValidationOnUnknownConnector() {
+        herder = exactlyOnceHerder();
+        Map<String, String> config = new HashMap<>();
+        config.put(SourceConnectorConfig.EXACTLY_ONCE_SUPPORT_CONFIG, REQUIRED.toString());
+
+        SourceConnector connectorMock = mock(SourceConnector.class);
+        when(connectorMock.exactlyOnceSupport(eq(config))).thenReturn(null);
+
+        Map<String, ConfigValue> validatedConfigs = herder.validateSourceConnectorConfig(
+                connectorMock, SourceConnectorConfig.configDef(), config);
+
+        List<String> errors = validatedConfigs.get(SourceConnectorConfig.EXACTLY_ONCE_SUPPORT_CONFIG).errorMessages();
+        assertFalse(errors.isEmpty());
+        assertTrue(
+                "Error message did not contain expected text: " + errors.get(0),
+                errors.get(0).contains("The connector does not implement the API required for preflight validation of exactly-once source support."));
+        assertEquals(1, errors.size());
+    }
+
+    @Test
+    public void testExactlyOnceSourceSupportValidationHandlesConnectorErrorsGracefully() {
+        herder = exactlyOnceHerder();
+        Map<String, String> config = new HashMap<>();
+        config.put(SourceConnectorConfig.EXACTLY_ONCE_SUPPORT_CONFIG, REQUIRED.toString());
+
+        SourceConnector connectorMock = mock(SourceConnector.class);
+        String errorMessage = "time to add a new unit test :)";
+        when(connectorMock.exactlyOnceSupport(eq(config))).thenThrow(new NullPointerException(errorMessage));
+
+        Map<String, ConfigValue> validatedConfigs = herder.validateSourceConnectorConfig(
+                connectorMock, SourceConnectorConfig.configDef(), config);
 
-        PowerMock.verifyAll();
+        List<String> errors = validatedConfigs.get(SourceConnectorConfig.EXACTLY_ONCE_SUPPORT_CONFIG).errorMessages();
+        assertFalse(errors.isEmpty());
+        assertTrue(
+                "Error message did not contain expected text: " + errors.get(0),
+                errors.get(0).contains(errorMessage));
+        assertEquals(1, errors.size());
     }
 
     @Test
-    public void testTaskReconfigurationRetriesWithConnectorTaskConfigsException() {
-        EasyMock.expect(member.memberId()).andStubReturn("leader");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
-        expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true);
-        expectConfigRefreshAndSnapshot(SNAPSHOT);
+    public void testExactlyOnceSourceSupportValidationWhenExactlyOnceNotEnabledOnWorker() {
+        Map<String, String> config = new HashMap<>();
+        config.put(SourceConnectorConfig.EXACTLY_ONCE_SUPPORT_CONFIG, REQUIRED.toString());
 
-        // end of initial tick
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
+        SourceConnector connectorMock = mock(SourceConnector.class);
+        when(connectorMock.exactlyOnceSupport(eq(config))).thenReturn(ExactlyOnceSupport.SUPPORTED);
 
-        member.wakeup();
-        PowerMock.expectLastCall().anyTimes();
+        Map<String, ConfigValue> validatedConfigs = herder.validateSourceConnectorConfig(
+                connectorMock, SourceConnectorConfig.configDef(), config);
 
-        member.ensureActive();
-        PowerMock.expectLastCall().anyTimes();
+        List<String> errors = validatedConfigs.get(SourceConnectorConfig.EXACTLY_ONCE_SUPPORT_CONFIG).errorMessages();
+        assertEquals(
+                Collections.singletonList("This worker does not have exactly-once source support enabled."),
+                errors);
+    }
 
-        EasyMock.expect(worker.isRunning(CONN1)).andReturn(true).anyTimes();
-        EasyMock.expect(worker.getPlugins()).andReturn(plugins).anyTimes();
+    @Test
+    public void testExactlyOnceSourceSupportValidationHandlesInvalidValuesGracefully() {
+        herder = exactlyOnceHerder();
+        Map<String, String> config = new HashMap<>();
+        config.put(SourceConnectorConfig.EXACTLY_ONCE_SUPPORT_CONFIG, "invalid");
 
-        SinkConnectorConfig sinkConnectorConfig = new SinkConnectorConfig(plugins, CONN1_CONFIG);
-        EasyMock.expect(worker.connectorTaskConfigs(CONN1, sinkConnectorConfig))
-                .andThrow(new ConnectException("Failed to generate task configs")).times(2);
+        SourceConnector connectorMock = mock(SourceConnector.class);
 
-        EasyMock.expect(worker.connectorTaskConfigs(CONN1, sinkConnectorConfig)).andReturn(TASK_CONFIGS);
+        Map<String, ConfigValue> validatedConfigs = herder.validateSourceConnectorConfig(
+                connectorMock, SourceConnectorConfig.configDef(), config);
 
-        expectAndVerifyTaskReconfigurationRetries();
+        List<String> errors = validatedConfigs.get(SourceConnectorConfig.EXACTLY_ONCE_SUPPORT_CONFIG).errorMessages();
+        assertFalse(errors.isEmpty());
+        assertTrue(
+                "Error message did not contain expected text: " + errors.get(0),
+                errors.get(0).contains("String must be one of (case insensitive): "));
+        assertEquals(1, errors.size());
     }
 
     @Test
-    public void testTaskReconfigurationRetriesWithLeaderRequestForwardingException() {
-        herder = PowerMock.createPartialMock(DistributedHerder.class,
-                new String[]{"connectorType", "updateDeletedConnectorStatus", "updateDeletedTaskStatus", "validateConnectorConfig", "buildRestartPlan", "recordRestarting"},
-                new DistributedConfig(HERDER_CONFIG), worker, WORKER_ID, KAFKA_CLUSTER_ID,
-                statusBackingStore, configBackingStore, member, MEMBER_URL, restClient, metrics, time, noneConnectorClientConfigOverridePolicy,
-                Collections.emptyList(), new MockSynchronousExecutor(), new AutoCloseable[]{});
+    public void testConnectorTransactionBoundaryValidation() {
+        herder = exactlyOnceHerder();
+        Map<String, String> config = new HashMap<>();
+        config.put(SourceConnectorConfig.TRANSACTION_BOUNDARY_CONFIG, CONNECTOR.toString());
 
-        rebalanceListener = herder.new RebalanceListener(time);
+        SourceConnector connectorMock = mock(SourceConnector.class);
+        when(connectorMock.canDefineTransactionBoundaries(eq(config)))
+                .thenReturn(ConnectorTransactionBoundaries.SUPPORTED);
 
-        EasyMock.expect(member.memberId()).andStubReturn("member");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
-        expectRebalance(1, Collections.emptyList(), Collections.emptyList(), false);
-        expectConfigRefreshAndSnapshot(SNAPSHOT);
+        Map<String, ConfigValue> validatedConfigs = herder.validateSourceConnectorConfig(
+                connectorMock, SourceConnectorConfig.configDef(), config);
 
-        // end of initial tick
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
+        List<String> errors = validatedConfigs.get(SourceConnectorConfig.TRANSACTION_BOUNDARY_CONFIG).errorMessages();
+        assertEquals(Collections.emptyList(), errors);
+    }
 
-        member.wakeup();
-        PowerMock.expectLastCall().anyTimes();
+    @Test
+    public void testConnectorTransactionBoundaryValidationOnUnsupportedConnector() {
+        herder = exactlyOnceHerder();
+        Map<String, String> config = new HashMap<>();
+        config.put(SourceConnectorConfig.TRANSACTION_BOUNDARY_CONFIG, CONNECTOR.toString());
 
-        member.ensureActive();
-        PowerMock.expectLastCall().anyTimes();
+        SourceConnector connectorMock = mock(SourceConnector.class);
+        when(connectorMock.canDefineTransactionBoundaries(eq(config)))
+                .thenReturn(ConnectorTransactionBoundaries.UNSUPPORTED);
 
-        EasyMock.expect(worker.isRunning(CONN1)).andReturn(true).anyTimes();
-        EasyMock.expect(worker.getPlugins()).andReturn(plugins).anyTimes();
+        Map<String, ConfigValue> validatedConfigs = herder.validateSourceConnectorConfig(
+                connectorMock, SourceConnectorConfig.configDef(), config);
 
-        SinkConnectorConfig sinkConnectorConfig = new SinkConnectorConfig(plugins, CONN1_CONFIG);
+        List<String> errors = validatedConfigs.get(SourceConnectorConfig.TRANSACTION_BOUNDARY_CONFIG).errorMessages();
+        assertFalse(errors.isEmpty());
+        assertTrue(
+                "Error message did not contain expected text: " + errors.get(0),
+                errors.get(0).contains("The connector does not support connector-defined transaction boundaries with the given configuration."));
+        assertEquals(1, errors.size());
+    }
 
-        List<Map<String, String>> changedTaskConfigs = new ArrayList<>(TASK_CONFIGS);
-        changedTaskConfigs.add(TASK_CONFIG);
-        EasyMock.expect(worker.connectorTaskConfigs(CONN1, sinkConnectorConfig)).andReturn(changedTaskConfigs).anyTimes();
+    @Test
+    public void testConnectorTransactionBoundaryValidationHandlesConnectorErrorsGracefully() {
+        herder = exactlyOnceHerder();
+        Map<String, String> config = new HashMap<>();
+        config.put(SourceConnectorConfig.TRANSACTION_BOUNDARY_CONFIG, CONNECTOR.toString());
 
-        EasyMock.expect(restClient.httpRequest(
-                EasyMock.anyString(), EasyMock.eq("POST"), EasyMock.anyObject(HttpHeaders.class),
-                EasyMock.anyObject(), EasyMock.anyObject(), EasyMock.anyObject(SecretKey.class), EasyMock.anyString())
-        ).andThrow(new ConnectException("Request to leader to reconfigure connector tasks failed")).times(2);
+        SourceConnector connectorMock = mock(SourceConnector.class);
+        String errorMessage = "Wait I thought we tested for this?";
+        when(connectorMock.canDefineTransactionBoundaries(eq(config))).thenThrow(new ConnectException(errorMessage));
 
-        EasyMock.expect(restClient.httpRequest(
-                EasyMock.anyString(), EasyMock.eq("POST"), EasyMock.anyObject(HttpHeaders.class),
-                EasyMock.anyObject(), EasyMock.anyObject(), EasyMock.anyObject(SecretKey.class), EasyMock.anyString())
-        ).andReturn(null);
+        Map<String, ConfigValue> validatedConfigs = herder.validateSourceConnectorConfig(
+                connectorMock, SourceConnectorConfig.configDef(), config);
 
-        expectAndVerifyTaskReconfigurationRetries();
+        List<String> errors = validatedConfigs.get(SourceConnectorConfig.TRANSACTION_BOUNDARY_CONFIG).errorMessages();
+        assertFalse(errors.isEmpty());
+        assertTrue(
+                "Error message did not contain expected text: " + errors.get(0),
+                errors.get(0).contains(errorMessage));
+        assertEquals(1, errors.size());
     }
 
-    private void expectAndVerifyTaskReconfigurationRetries() {
-        // task reconfiguration herder request with initial retry backoff
-        member.poll(EasyMock.eq(250L));
-        PowerMock.expectLastCall();
-
-        // task reconfiguration herder request with double the initial retry backoff
-        member.poll(EasyMock.eq(500L));
-        PowerMock.expectLastCall();
-
-        // the third task reconfiguration request is expected to pass; so expect no more retries (a Long.MAX_VALUE poll
-        // timeout indicates that there is no herder request currently in the queue)
-        member.poll(EasyMock.eq(Long.MAX_VALUE));
-        PowerMock.expectLastCall();
+    @Test
+    public void testConnectorTransactionBoundaryValidationHandlesInvalidValuesGracefully() {
+        herder = exactlyOnceHerder();
+        Map<String, String> config = new HashMap<>();
+        config.put(SourceConnectorConfig.TRANSACTION_BOUNDARY_CONFIG, "CONNECTOR.toString()");
 
-        PowerMock.replayAll();
+        SourceConnector connectorMock = mock(SourceConnector.class);
 
-        // initial tick
-        herder.tick();
-        herder.requestTaskReconfiguration(CONN1);
-        // process the task reconfiguration request in this tick
-        herder.tick();
-        // advance the time by 250ms so that the task reconfiguration request with initial retry backoff is processed
-        time.sleep(250);
-        herder.tick();
-        // advance the time by 500ms so that the task reconfiguration request with double the initial retry backoff is processed
-        time.sleep(500);
-        herder.tick();
+        Map<String, ConfigValue> validatedConfigs = herder.validateSourceConnectorConfig(
+                connectorMock, SourceConnectorConfig.configDef(), config);
 
-        PowerMock.verifyAll();
+        List<String> errors = validatedConfigs.get(SourceConnectorConfig.TRANSACTION_BOUNDARY_CONFIG).errorMessages();
+        assertFalse(errors.isEmpty());
+        assertTrue(
+                "Error message did not contain expected text: " + errors.get(0),
+                errors.get(0).contains("String must be one of (case insensitive): "));
+        assertEquals(1, errors.size());
     }
 
     @Test
     public void testConnectorOffsets() throws Exception {
-        EasyMock.expect(member.memberId()).andStubReturn("leader");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+        when(member.memberId()).thenReturn("leader");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
         expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true);
         expectConfigRefreshAndSnapshot(SNAPSHOT);
-        expectAnyTicks();
 
-        member.wakeup();
-        PowerMock.expectLastCall();
+        when(statusBackingStore.connectors()).thenReturn(Collections.emptySet());
+        doNothing().when(member).poll(anyLong());
 
-        EasyMock.expect(configBackingStore.snapshot()).andReturn(SNAPSHOT);
+        herder.tick();
+
+        when(configBackingStore.snapshot()).thenReturn(SNAPSHOT);
         ConnectorOffsets offsets = new ConnectorOffsets(Collections.singletonList(new ConnectorOffset(
                 Collections.singletonMap("partitionKey", "partitionValue"),
                 Collections.singletonMap("offsetKey", "offsetValue"))));
-        Capture<Callback<ConnectorOffsets>> callbackCapture = newCapture();
-        worker.connectorOffsets(EasyMock.eq(CONN1), EasyMock.eq(CONN1_CONFIG), capture(callbackCapture));
-        PowerMock.expectLastCall().andAnswer(() -> {
+
+        ArgumentCaptor<Callback<ConnectorOffsets>> callbackCapture = ArgumentCaptor.forClass(Callback.class);
+        doAnswer(invocation -> {
             callbackCapture.getValue().onCompletion(null, offsets);
             return null;
-        });
+        }).when(worker).connectorOffsets(eq(CONN1), eq(CONN1_CONFIG), callbackCapture.capture());
 
-        PowerMock.replayAll();
-
-        herder.tick();
         FutureCallback<ConnectorOffsets> cb = new FutureCallback<>();
         herder.connectorOffsets(CONN1, cb);
         herder.tick();
         assertEquals(offsets, cb.get(1000, TimeUnit.MILLISECONDS));
 
-        PowerMock.verifyAll();
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testModifyConnectorOffsetsUnknownConnector() throws Exception {
         // Get the initial assignment
-        EasyMock.expect(member.memberId()).andStubReturn("leader");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+        when(member.memberId()).thenReturn("leader");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
         expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true);
         expectConfigRefreshAndSnapshot(SNAPSHOT);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
-
-        // Now handle the connector offsets modification request
-        member.wakeup();
-        PowerMock.expectLastCall();
-        member.ensureActive();
-        PowerMock.expectLastCall();
-        expectConfigRefreshAndSnapshot(SNAPSHOT);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
-
-        PowerMock.replayAll();
 
         herder.tick();
+
+        // Now handle the connector offsets modification request
         FutureCallback<Message> callback = new FutureCallback<>();
         herder.modifyConnectorOffsets("connector-does-not-exist", new HashMap<>(), callback);
         herder.tick();
         ExecutionException e = assertThrows(ExecutionException.class, () -> callback.get(1000L, TimeUnit.MILLISECONDS));
         assertTrue(e.getCause() instanceof NotFoundException);
-
-        PowerMock.verifyAll();
     }
 
     @Test
     public void testModifyOffsetsConnectorNotInStoppedState() throws Exception {
         // Get the initial assignment
-        EasyMock.expect(member.memberId()).andStubReturn("leader");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+        when(member.memberId()).thenReturn("leader");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
         expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true);
         expectConfigRefreshAndSnapshot(SNAPSHOT);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
-
-        // Now handle the connector offsets modification request
-        member.wakeup();
-        PowerMock.expectLastCall();
-        member.ensureActive();
-        PowerMock.expectLastCall();
-        expectConfigRefreshAndSnapshot(SNAPSHOT);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
-
-        PowerMock.replayAll();
 
         herder.tick();
+
+        // Now handle the connector offsets modification request
         FutureCallback<Message> callback = new FutureCallback<>();
         herder.modifyConnectorOffsets(CONN1, null, callback);
         herder.tick();
         ExecutionException e = assertThrows(ExecutionException.class, () -> callback.get(1000L, TimeUnit.MILLISECONDS));
         assertTrue(e.getCause() instanceof BadRequestException);
-
-        PowerMock.verifyAll();
     }
 
     @Test
     public void testModifyOffsetsNotLeader() throws Exception {
         // Get the initial assignment
-        EasyMock.expect(member.memberId()).andStubReturn("member");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+        when(member.memberId()).thenReturn("member");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
         expectRebalance(1, Collections.emptyList(), Collections.emptyList(), false);
         expectConfigRefreshAndSnapshot(SNAPSHOT_STOPPED_CONN1);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
-
-        // Now handle the connector offsets modification request
-        member.wakeup();
-        PowerMock.expectLastCall();
-        member.ensureActive();
-        PowerMock.expectLastCall();
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
-
-        PowerMock.replayAll();
 
         herder.tick();
+
+        // Now handle the connector offsets modification request
         FutureCallback<Message> callback = new FutureCallback<>();
         herder.modifyConnectorOffsets(CONN1, new HashMap<>(), callback);
         herder.tick();
         ExecutionException e = assertThrows(ExecutionException.class, () -> callback.get(1000L, TimeUnit.MILLISECONDS));
         assertTrue(e.getCause() instanceof NotLeaderException);
-
-        PowerMock.verifyAll();
     }
 
     @Test
     public void testModifyOffsetsSinkConnector() throws Exception {
-        EasyMock.reset(herder);
-        EasyMock.expect(herder.connectorType(EasyMock.anyObject())).andReturn(ConnectorType.SINK).anyTimes();
-        PowerMock.expectPrivate(herder, "updateDeletedConnectorStatus").andVoid().anyTimes();
-        PowerMock.expectPrivate(herder, "updateDeletedTaskStatus").andVoid().anyTimes();
-
+        when(herder.connectorType(any())).thenReturn(ConnectorType.SINK);
         // Get the initial assignment
-        EasyMock.expect(member.memberId()).andStubReturn("leader");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+        when(member.memberId()).thenReturn("leader");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
         expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true);
         expectConfigRefreshAndSnapshot(SNAPSHOT_STOPPED_CONN1);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
+
+        herder.tick();
 
         // Now handle the alter connector offsets request
         Map<Map<String, ?>, Map<String, ?>> offsets = Collections.singletonMap(
                 Collections.singletonMap("partitionKey", "partitionValue"),
                 Collections.singletonMap("offsetKey", "offsetValue"));
-        member.wakeup();
-        PowerMock.expectLastCall();
-        member.ensureActive();
-        PowerMock.expectLastCall();
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
-        expectConfigRefreshAndSnapshot(SNAPSHOT_STOPPED_CONN1);
-        Capture<Callback<Message>> workerCallbackCapture = Capture.newInstance();
-        worker.modifyConnectorOffsets(EasyMock.eq(CONN1), EasyMock.eq(CONN1_CONFIG), EasyMock.eq(offsets), capture(workerCallbackCapture));
+
+        ArgumentCaptor<Callback<Message>> workerCallbackCapture = ArgumentCaptor.forClass(Callback.class);
         Message msg = new Message("The offsets for this connector have been altered successfully");
-        EasyMock.expectLastCall().andAnswer(() -> {
+        doAnswer(invocation -> {
             workerCallbackCapture.getValue().onCompletion(null, msg);
             return null;
-        });
+        }).when(worker).modifyConnectorOffsets(eq(CONN1), eq(CONN1_CONFIG), eq(offsets), workerCallbackCapture.capture());
 
-        PowerMock.replayAll();
-
-        herder.tick();
         FutureCallback<Message> callback = new FutureCallback<>();
         herder.alterConnectorOffsets(CONN1, offsets, callback);
         herder.tick();
         assertEquals(msg, callback.get(1000L, TimeUnit.MILLISECONDS));
         assertEquals("The offsets for this connector have been altered successfully", msg.message());
-
-        PowerMock.verifyAll();
     }
 
     @Test
     public void testModifyOffsetsSourceConnectorExactlyOnceDisabled() throws Exception {
         // Get the initial assignment
-        EasyMock.expect(member.memberId()).andStubReturn("leader");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+        when(member.memberId()).thenReturn("leader");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
         expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true);
         expectConfigRefreshAndSnapshot(SNAPSHOT_STOPPED_CONN1);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
+        herder.tick();
 
         // Now handle the reset connector offsets request
-        member.wakeup();
-        PowerMock.expectLastCall();
-        member.ensureActive();
-        PowerMock.expectLastCall();
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
-        expectConfigRefreshAndSnapshot(SNAPSHOT_STOPPED_CONN1);
-        Capture<Callback<Message>> workerCallbackCapture = Capture.newInstance();
-        worker.modifyConnectorOffsets(EasyMock.eq(CONN1), EasyMock.eq(CONN1_CONFIG), isNull(), capture(workerCallbackCapture));
+        ArgumentCaptor<Callback<Message>> workerCallbackCapture = ArgumentCaptor.forClass(Callback.class);
         Message msg = new Message("The offsets for this connector have been reset successfully");
-        EasyMock.expectLastCall().andAnswer(() -> {
+        doAnswer(invocation -> {
             workerCallbackCapture.getValue().onCompletion(null, msg);
             return null;
-        });
+        }).when(worker).modifyConnectorOffsets(eq(CONN1), eq(CONN1_CONFIG), isNull(), workerCallbackCapture.capture());
 
-        PowerMock.replayAll();
-
-        herder.tick();
         FutureCallback<Message> callback = new FutureCallback<>();
         herder.resetConnectorOffsets(CONN1, callback);
         herder.tick();
         assertEquals(msg, callback.get(1000L, TimeUnit.MILLISECONDS));
         assertEquals("The offsets for this connector have been reset successfully", msg.message());
-
-        PowerMock.verifyAll();
     }
 
     @Test
     public void testModifyOffsetsSourceConnectorExactlyOnceEnabled() throws Exception {
         // Setup herder with exactly-once support for source connectors enabled
         herder = exactlyOnceHerder();
         rebalanceListener = herder.new RebalanceListener(time);
-        PowerMock.expectPrivate(herder, "updateDeletedConnectorStatus").andVoid().anyTimes();
-        PowerMock.expectPrivate(herder, "updateDeletedTaskStatus").andVoid().anyTimes();
-
         // Get the initial assignment
-        EasyMock.expect(member.memberId()).andStubReturn("leader");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+        when(member.memberId()).thenReturn("leader");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
         expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true);
         expectConfigRefreshAndSnapshot(SNAPSHOT_STOPPED_CONN1);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall().anyTimes();
+        doNothing().when(member).poll(anyLong());
+
+        herder.tick();
 
         // Now handle the alter connector offsets request
-        Map<Map<String, ?>, Map<String, ?>> offsets = Collections.singletonMap(
-                Collections.singletonMap("partitionKey", "partitionValue"),
-                Collections.singletonMap("offsetKey", "offsetValue"));
-        member.wakeup();
-        PowerMock.expectLastCall().anyTimes();
-        member.ensureActive();
-        PowerMock.expectLastCall().anyTimes();
-        expectConfigRefreshAndSnapshot(SNAPSHOT_STOPPED_CONN1);
-        expectConfigRefreshAndSnapshot(SNAPSHOT_STOPPED_CONN1);
-        EasyMock.expect(herder.connectorType(EasyMock.anyObject())).andReturn(ConnectorType.SOURCE).anyTimes();
+        doNothing().when(member).ensureActive();
+        when(herder.connectorType(any())).thenReturn(ConnectorType.SOURCE);
 
         // Expect a round of zombie fencing to occur
-        expectConfigRefreshAndSnapshot(SNAPSHOT_STOPPED_CONN1);
-        KafkaFuture<Void> workerFencingFuture = EasyMock.mock(KafkaFuture.class);
-        KafkaFuture<Void> herderFencingFuture = EasyMock.mock(KafkaFuture.class);
-        EasyMock.expect(worker.fenceZombies(CONN1, SNAPSHOT_STOPPED_CONN1.taskCountRecord(CONN1), CONN1_CONFIG)).andReturn(workerFencingFuture);
-        EasyMock.expect(workerFencingFuture.thenApply(EasyMock.<KafkaFuture.BaseFunction<Void, Void>>anyObject())).andReturn(herderFencingFuture);
-
-        // Two fencing callbacks are added - one is in ZombieFencing::start itself to remove the connector from the active
-        // fencing list. The other is the callback passed from DistributedHerder::modifyConnectorOffsets in order to
-        // queue up the actual alter offsets request if the zombie fencing succeeds.
-        for (int i = 0; i < 2; i++) {
-            Capture<KafkaFuture.BiConsumer<Void, Throwable>> herderFencingCallback = EasyMock.newCapture();
-            EasyMock.expect(herderFencingFuture.whenComplete(EasyMock.capture(herderFencingCallback))).andAnswer(() -> {
-                herderFencingCallback.getValue().accept(null, null);
-                return null;
-            });
-        }
-
-        Capture<Callback<Message>> workerCallbackCapture = Capture.newInstance();
-        Message msg = new Message("The offsets for this connector have been altered successfully");
-        worker.modifyConnectorOffsets(EasyMock.eq(CONN1), EasyMock.eq(CONN1_CONFIG), EasyMock.eq(offsets), capture(workerCallbackCapture));
-        EasyMock.expectLastCall().andAnswer(() -> {
-            workerCallbackCapture.getValue().onCompletion(null, msg);
+        KafkaFuture<Void> workerFencingFuture = mock(KafkaFuture.class);
+        KafkaFuture<Void> herderFencingFuture = mock(KafkaFuture.class);
+        when(worker.fenceZombies(CONN1, SNAPSHOT_STOPPED_CONN1.taskCountRecord(CONN1), CONN1_CONFIG)).thenReturn(workerFencingFuture);
+        when(workerFencingFuture.thenApply(any(KafkaFuture.BaseFunction.class))).thenReturn(herderFencingFuture);
+
+        ArgumentCaptor<KafkaFuture.BiConsumer<Void, Throwable>> herderFencingCallback = ArgumentCaptor.forClass(KafkaFuture.BiConsumer.class);
+        when(herderFencingFuture.whenComplete(herderFencingCallback.capture())).thenAnswer(invocation -> {
+            herderFencingCallback.getValue().accept(null, null);
             return null;
         });
 
-        // Handle the second alter connector offsets request. No zombie fencing request to the worker is expected now since we
-        // already did a round of zombie fencing last time and no new tasks came up in the meanwhile. The config snapshot is
-        // refreshed once at the beginning of the DistributedHerder::modifyConnectorOffsets method, once before checking
-        // whether zombie fencing is required, and once before actually proceeding to alter connector offsets.
-        expectConfigRefreshAndSnapshot(SNAPSHOT_STOPPED_CONN1_FENCED);
-        expectConfigRefreshAndSnapshot(SNAPSHOT_STOPPED_CONN1_FENCED);
-        expectConfigRefreshAndSnapshot(SNAPSHOT_STOPPED_CONN1_FENCED);
-        Capture<Callback<Message>> workerCallbackCapture2 = Capture.newInstance();
-        worker.modifyConnectorOffsets(EasyMock.eq(CONN1), EasyMock.eq(CONN1_CONFIG), EasyMock.eq(offsets), capture(workerCallbackCapture2));
-        EasyMock.expectLastCall().andAnswer(() -> {
-            workerCallbackCapture2.getValue().onCompletion(null, msg);
-            return null;
-        });
+        ArgumentCaptor<Callback<Message>> workerCallbackCapture = ArgumentCaptor.forClass(Callback.class);
+        Message msg = new Message("The offsets for this connector have been altered successfully");
 
-        PowerMock.replayAll(workerFencingFuture, herderFencingFuture);
+        Map<Map<String, ?>, Map<String, ?>> offsets = Collections.singletonMap(
+                Collections.singletonMap("partitionKey", "partitionValue"),
+                Collections.singletonMap("offsetKey", "offsetValue"));
+        doAnswer(invocation -> {
+            workerCallbackCapture.getValue().onCompletion(null, msg);
+            return null;
+        }).when(worker).modifyConnectorOffsets(eq(CONN1), eq(CONN1_CONFIG), eq(offsets), workerCallbackCapture.capture());
 
-        herder.tick();
         FutureCallback<Message> callback = new FutureCallback<>();
         herder.alterConnectorOffsets(CONN1, offsets, callback);
-        // Process the zombie fencing request
-        herder.tick();
-        // Process the alter offsets request
+        // Process the zombie fencing request that is queued up first followed by the actual alter offsets request

Review Comment:
   > Ah... is it because we're adding the followup request to alter the connector offsets to the herder's request queue while it's in the middle of handling the initial request to fence zombie source tasks, and so the herder immediately handles the follow-up request once it's done with the first one?
   
   Yep, exactly. This bit will basically run synchronously -
   https://github.com/apache/kafka/blob/f23394336a7741bf4eb23fcde951af0a23a69bd0/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java#L1536-L1547
   
   which will result in the modify connector offsets request being run in the same request processing loop as the zombie fencing request here - 
   https://github.com/apache/kafka/blob/f23394336a7741bf4eb23fcde951af0a23a69bd0/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java#L457-L469
   
   
   In fact the original test also passes without the second tick call.



##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java:
##########
@@ -1466,560 +1032,421 @@ public void testRestartConnectorAndTasksUnknownConnector() throws Exception {
         assertTrue(ee.getCause() instanceof NotFoundException);
         assertTrue(ee.getMessage().contains("Unknown connector:"));
 
-        PowerMock.verifyAll();
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testRestartConnectorAndTasksNotLeader() throws Exception {
-        RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
-
         // get the initial assignment
-        EasyMock.expect(member.memberId()).andStubReturn("member");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+        when(member.memberId()).thenReturn("member");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
         expectRebalance(1, Collections.emptyList(), Collections.emptyList());
         expectConfigRefreshAndSnapshot(SNAPSHOT);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
+        doNothing().when(member).poll(anyLong());
 
-        // now handle the connector restart
-        member.wakeup();
-        PowerMock.expectLastCall();
-        member.ensureActive();
-        PowerMock.expectLastCall();
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
+        herder.tick();
 
-        PowerMock.replayAll();
+        // now handle the connector restart
+        doNothing().when(member).ensureActive();
 
         herder.tick();
         FutureCallback<ConnectorStateInfo> callback = new FutureCallback<>();
+        RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
         herder.restartConnectorAndTasks(restartRequest, callback);
         herder.tick();
         ExecutionException ee = assertThrows(ExecutionException.class, () -> callback.get(1000L, TimeUnit.MILLISECONDS));
         assertTrue(ee.getCause() instanceof NotLeaderException);
 
-        PowerMock.verifyAll();
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testRestartConnectorAndTasksUnknownStatus() throws Exception {
-        RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
-        EasyMock.expect(herder.buildRestartPlan(restartRequest)).andReturn(Optional.empty()).anyTimes();
-
-        configBackingStore.putRestartRequest(restartRequest);
-        PowerMock.expectLastCall();
-
         // get the initial assignment
-        EasyMock.expect(member.memberId()).andStubReturn("leader");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+        when(member.memberId()).thenReturn("leader");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
         expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true);
         expectConfigRefreshAndSnapshot(SNAPSHOT);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
-
-        // now handle the connector restart
-        member.wakeup();
-        PowerMock.expectLastCall();
-        member.ensureActive();
-        PowerMock.expectLastCall();
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
 
-        PowerMock.replayAll();
+        when(statusBackingStore.connectors()).thenReturn(Collections.emptySet());
+        doNothing().when(member).poll(anyLong());
 
         herder.tick();
+
+        // now handle the connector restart
+        doNothing().when(member).ensureActive();
+        when(statusBackingStore.get(CONN1)).thenReturn(null);
+        RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
+        doNothing().when(configBackingStore).putRestartRequest(restartRequest);
+
         FutureCallback<ConnectorStateInfo> callback = new FutureCallback<>();
         herder.restartConnectorAndTasks(restartRequest, callback);
         herder.tick();
         ExecutionException ee = assertThrows(ExecutionException.class, () -> callback.get(1000L, TimeUnit.MILLISECONDS));
         assertTrue(ee.getCause() instanceof NotFoundException);
         assertTrue(ee.getMessage().contains("Status for connector"));
-        PowerMock.verifyAll();
+
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testRestartConnectorAndTasksSuccess() throws Exception {
-        RestartPlan restartPlan = PowerMock.createMock(RestartPlan.class);
-        ConnectorStateInfo connectorStateInfo = PowerMock.createMock(ConnectorStateInfo.class);
-        EasyMock.expect(restartPlan.restartConnectorStateInfo()).andReturn(connectorStateInfo).anyTimes();
-
-        RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
-        EasyMock.expect(herder.buildRestartPlan(restartRequest)).andReturn(Optional.of(restartPlan)).anyTimes();
-
-        configBackingStore.putRestartRequest(restartRequest);
-        PowerMock.expectLastCall();
-
         // get the initial assignment
-        EasyMock.expect(member.memberId()).andStubReturn("leader");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+        when(member.memberId()).thenReturn("leader");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
         expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true);
         expectConfigRefreshAndSnapshot(SNAPSHOT);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
+
+        when(statusBackingStore.connectors()).thenReturn(Collections.emptySet());
+        doNothing().when(member).poll(anyLong());
+
+        herder.tick();
 
         // now handle the connector restart
-        member.wakeup();
-        PowerMock.expectLastCall();
-        member.ensureActive();
-        PowerMock.expectLastCall();
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
+        doNothing().when(member).ensureActive();
 
-        PowerMock.replayAll();
+        RestartPlan restartPlan = mock(RestartPlan.class);
+        ConnectorStateInfo connectorStateInfo = mock(ConnectorStateInfo.class);
+        when(restartPlan.restartConnectorStateInfo()).thenReturn(connectorStateInfo);
+        RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
+        doReturn(Optional.of(restartPlan)).when(herder).buildRestartPlan(restartRequest);
+        doNothing().when(configBackingStore).putRestartRequest(restartRequest);
 
-        herder.tick();
         FutureCallback<ConnectorStateInfo> callback = new FutureCallback<>();
         herder.restartConnectorAndTasks(restartRequest, callback);
         herder.tick();
         assertEquals(connectorStateInfo,  callback.get(1000L, TimeUnit.MILLISECONDS));
-        PowerMock.verifyAll();
+
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testDoRestartConnectorAndTasksEmptyPlan() {
         RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
-        EasyMock.expect(herder.buildRestartPlan(restartRequest)).andReturn(Optional.empty()).anyTimes();
-
-        PowerMock.replayAll();
-
+        doReturn(Optional.empty()).when(herder).buildRestartPlan(restartRequest);
         herder.doRestartConnectorAndTasks(restartRequest);
-        PowerMock.verifyAll();
+
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testDoRestartConnectorAndTasksNoAssignments() {
         ConnectorTaskId taskId = new ConnectorTaskId(CONN1, 0);
         RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
-        RestartPlan restartPlan = PowerMock.createMock(RestartPlan.class);
-        EasyMock.expect(restartPlan.shouldRestartConnector()).andReturn(true).anyTimes();
-        EasyMock.expect(restartPlan.shouldRestartTasks()).andReturn(true).anyTimes();
-        EasyMock.expect(restartPlan.taskIdsToRestart()).andReturn(Collections.singletonList(taskId)).anyTimes();
+        RestartPlan restartPlan = mock(RestartPlan.class);
+        when(restartPlan.shouldRestartConnector()).thenReturn(true);
+        when(restartPlan.taskIdsToRestart()).thenReturn(Collections.singletonList(taskId));
 
-        EasyMock.expect(herder.buildRestartPlan(restartRequest)).andReturn(Optional.of(restartPlan)).anyTimes();
+        doReturn(Optional.of(restartPlan)).when(herder).buildRestartPlan(restartRequest);
 
-        PowerMock.replayAll();
         herder.assignment = ExtendedAssignment.empty();
         herder.doRestartConnectorAndTasks(restartRequest);
-        PowerMock.verifyAll();
+
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testDoRestartConnectorAndTasksOnlyConnector() {
         ConnectorTaskId taskId = new ConnectorTaskId(CONN1, 0);
         RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
-        RestartPlan restartPlan = PowerMock.createMock(RestartPlan.class);
-        EasyMock.expect(restartPlan.shouldRestartConnector()).andReturn(true).anyTimes();
-        EasyMock.expect(restartPlan.shouldRestartTasks()).andReturn(true).anyTimes();
-        EasyMock.expect(restartPlan.taskIdsToRestart()).andReturn(Collections.singletonList(taskId)).anyTimes();
+        RestartPlan restartPlan = mock(RestartPlan.class);
+        when(restartPlan.shouldRestartConnector()).thenReturn(true);
+        when(restartPlan.taskIdsToRestart()).thenReturn(Collections.singletonList(taskId));
 
-        EasyMock.expect(herder.buildRestartPlan(restartRequest)).andReturn(Optional.of(restartPlan)).anyTimes();
+        doReturn(Optional.of(restartPlan)).when(herder).buildRestartPlan(restartRequest);
 
-        herder.assignment = PowerMock.createMock(ExtendedAssignment.class);
-        EasyMock.expect(herder.assignment.connectors()).andReturn(Collections.singletonList(CONN1)).anyTimes();
-        EasyMock.expect(herder.assignment.tasks()).andReturn(Collections.emptyList()).anyTimes();
+        herder.assignment = mock(ExtendedAssignment.class);
+        when(herder.assignment.connectors()).thenReturn(Collections.singletonList(CONN1));
+        when(herder.assignment.tasks()).thenReturn(Collections.emptyList());
 
-        worker.stopAndAwaitConnector(CONN1);
-        PowerMock.expectLastCall();
+        doNothing().when(worker).stopAndAwaitConnector(CONN1);
 
-        Capture<Callback<TargetState>>  stateCallback = newCapture();
-        worker.startConnector(EasyMock.eq(CONN1), EasyMock.anyObject(), EasyMock.anyObject(),
-                EasyMock.eq(herder), EasyMock.anyObject(TargetState.class), capture(stateCallback));
-        PowerMock.expectLastCall().andAnswer(() -> {
+        ArgumentCaptor<Callback<TargetState>>  stateCallback = ArgumentCaptor.forClass(Callback.class);
+        doAnswer(invocation -> {
             stateCallback.getValue().onCompletion(null, TargetState.STARTED);
             return true;
-        });
-        member.wakeup();
-        PowerMock.expectLastCall();
+        }).when(worker).startConnector(eq(CONN1), any(), any(), eq(herder), any(), stateCallback.capture());
+        doNothing().when(member).wakeup();
+        doNothing().when(statusBackingStore).put(any(ConnectorStatus.class));
 
-
-        herder.onRestart(CONN1);
-        EasyMock.expectLastCall();
-
-        PowerMock.replayAll();
         herder.doRestartConnectorAndTasks(restartRequest);
-        PowerMock.verifyAll();
+
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testDoRestartConnectorAndTasksOnlyTasks() {
         RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
-        RestartPlan restartPlan = PowerMock.createMock(RestartPlan.class);
-        EasyMock.expect(restartPlan.shouldRestartConnector()).andReturn(true).anyTimes();
-        EasyMock.expect(restartPlan.shouldRestartTasks()).andReturn(true).anyTimes();
+        RestartPlan restartPlan = mock(RestartPlan.class);
+        when(restartPlan.shouldRestartConnector()).thenReturn(true);
         // The connector has three tasks
-        EasyMock.expect(restartPlan.taskIdsToRestart()).andReturn(Arrays.asList(TASK0, TASK1, TASK2)).anyTimes();
-        EasyMock.expect(restartPlan.restartTaskCount()).andReturn(3).anyTimes();
-        EasyMock.expect(restartPlan.totalTaskCount()).andReturn(3).anyTimes();
-        EasyMock.expect(herder.buildRestartPlan(restartRequest)).andReturn(Optional.of(restartPlan)).anyTimes();
+        when(restartPlan.taskIdsToRestart()).thenReturn(Arrays.asList(TASK0, TASK1, TASK2));
+        when(restartPlan.totalTaskCount()).thenReturn(3);
+        doReturn(Optional.of(restartPlan)).when(herder).buildRestartPlan(restartRequest);
 
-        herder.assignment = PowerMock.createMock(ExtendedAssignment.class);
-        EasyMock.expect(herder.assignment.connectors()).andReturn(Collections.emptyList()).anyTimes();
+        herder.assignment = mock(ExtendedAssignment.class);
+        when(herder.assignment.connectors()).thenReturn(Collections.emptyList());
         // But only one task is assigned to this worker
-        EasyMock.expect(herder.assignment.tasks()).andReturn(Collections.singletonList(TASK0)).anyTimes();
+        when(herder.assignment.tasks()).thenReturn(Collections.singletonList(TASK0));
 
         herder.configState = SNAPSHOT;
 
-        worker.stopAndAwaitTasks(Collections.singletonList(TASK0));
-        PowerMock.expectLastCall();
-
-        herder.onRestart(TASK0);
-        EasyMock.expectLastCall();
-
-        worker.startSourceTask(EasyMock.eq(TASK0), EasyMock.anyObject(), EasyMock.anyObject(), EasyMock.anyObject(),
-                EasyMock.eq(herder), EasyMock.anyObject(TargetState.class));
-        PowerMock.expectLastCall().andReturn(true);
+        doNothing().when(worker).stopAndAwaitTasks(Collections.singletonList(TASK0));
+        doNothing().when(statusBackingStore).put(any(TaskStatus.class));
+        when(worker.startSourceTask(eq(TASK0), any(), any(), any(), eq(herder), any())).thenReturn(true);
 
-        PowerMock.replayAll();
         herder.doRestartConnectorAndTasks(restartRequest);
-        PowerMock.verifyAll();
+
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testDoRestartConnectorAndTasksBoth() {
         ConnectorTaskId taskId = new ConnectorTaskId(CONN1, 0);
         RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
-        RestartPlan restartPlan = PowerMock.createMock(RestartPlan.class);
-        EasyMock.expect(restartPlan.shouldRestartConnector()).andReturn(true).anyTimes();
-        EasyMock.expect(restartPlan.shouldRestartTasks()).andReturn(true).anyTimes();
-        EasyMock.expect(restartPlan.taskIdsToRestart()).andReturn(Collections.singletonList(taskId)).anyTimes();
-        EasyMock.expect(restartPlan.restartTaskCount()).andReturn(1).anyTimes();
-        EasyMock.expect(restartPlan.totalTaskCount()).andReturn(1).anyTimes();
-        EasyMock.expect(herder.buildRestartPlan(restartRequest)).andReturn(Optional.of(restartPlan)).anyTimes();
-
-        herder.assignment = PowerMock.createMock(ExtendedAssignment.class);
-        EasyMock.expect(herder.assignment.connectors()).andReturn(Collections.singletonList(CONN1)).anyTimes();
-        EasyMock.expect(herder.assignment.tasks()).andReturn(Collections.singletonList(taskId)).anyTimes();
-
-        herder.configState = SNAPSHOT;
+        RestartPlan restartPlan = mock(RestartPlan.class);
+        when(restartPlan.shouldRestartConnector()).thenReturn(true);
+        when(restartPlan.taskIdsToRestart()).thenReturn(Collections.singletonList(taskId));
+        when(restartPlan.totalTaskCount()).thenReturn(1);
+        doReturn(Optional.of(restartPlan)).when(herder).buildRestartPlan(restartRequest);
 
-        worker.stopAndAwaitConnector(CONN1);
-        PowerMock.expectLastCall();
-
-        Capture<Callback<TargetState>>  stateCallback = newCapture();
-        worker.startConnector(EasyMock.eq(CONN1), EasyMock.anyObject(), EasyMock.anyObject(),
-                EasyMock.eq(herder), EasyMock.anyObject(TargetState.class), capture(stateCallback));
+        herder.assignment = mock(ExtendedAssignment.class);
+        when(herder.assignment.connectors()).thenReturn(Collections.singletonList(CONN1));
+        when(herder.assignment.tasks()).thenReturn(Collections.singletonList(taskId));
 
+        herder.configState = SNAPSHOT;
 
-        herder.onRestart(CONN1);
-        EasyMock.expectLastCall();
+        doNothing().when(worker).stopAndAwaitConnector(CONN1);
 
-        worker.stopAndAwaitTasks(Collections.singletonList(taskId));
-        PowerMock.expectLastCall();
+        ArgumentCaptor<Callback<TargetState>>  stateCallback = ArgumentCaptor.forClass(Callback.class);
+        doAnswer(invocation -> {
+            stateCallback.getValue().onCompletion(null, TargetState.STARTED);
+            return true;
+        }).when(worker).startConnector(eq(CONN1), any(), any(), eq(herder), any(), stateCallback.capture());
+        doNothing().when(member).wakeup();
+        doNothing().when(statusBackingStore).put(any(ConnectorStatus.class));
 
-        herder.onRestart(taskId);
-        EasyMock.expectLastCall();
+        doNothing().when(worker).stopAndAwaitTasks(Collections.singletonList(taskId));
 
-        worker.startSourceTask(EasyMock.eq(TASK0), EasyMock.anyObject(), EasyMock.anyObject(), EasyMock.anyObject(),
-                EasyMock.eq(herder), EasyMock.anyObject(TargetState.class));
-        PowerMock.expectLastCall().andReturn(true);
+        when(worker.startSourceTask(eq(TASK0), any(), any(), any(), eq(herder), any())).thenReturn(true);
+        doNothing().when(statusBackingStore).put(any(TaskStatus.class));
 
-        PowerMock.replayAll();
         herder.doRestartConnectorAndTasks(restartRequest);
-        PowerMock.verifyAll();
+
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testRestartTask() throws Exception {
-        EasyMock.expect(worker.connectorTaskConfigs(CONN1, conn1SinkConfig)).andStubReturn(TASK_CONFIGS);
-
         // get the initial assignment
-        EasyMock.expect(member.memberId()).andStubReturn("leader");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+        when(member.memberId()).thenReturn("leader");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
         expectRebalance(1, Collections.emptyList(), singletonList(TASK0), true);
         expectConfigRefreshAndSnapshot(SNAPSHOT);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
-        worker.startSourceTask(EasyMock.eq(TASK0), EasyMock.anyObject(), EasyMock.anyObject(), EasyMock.anyObject(),
-                EasyMock.eq(herder), EasyMock.eq(TargetState.STARTED));
-        PowerMock.expectLastCall().andReturn(true);
-
-        // now handle the task restart
-        member.wakeup();
-        PowerMock.expectLastCall();
-        member.ensureActive();
-        PowerMock.expectLastCall();
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
 
-        worker.stopAndAwaitTask(TASK0);
-        PowerMock.expectLastCall();
-        worker.startSourceTask(EasyMock.eq(TASK0), EasyMock.anyObject(), EasyMock.anyObject(), EasyMock.anyObject(),
-                EasyMock.eq(herder), EasyMock.eq(TargetState.STARTED));
-        PowerMock.expectLastCall().andReturn(true);
+        when(statusBackingStore.connectors()).thenReturn(Collections.emptySet());
+        doNothing().when(member).poll(anyLong());
 
-        PowerMock.replayAll();
+        when(worker.startSourceTask(eq(TASK0), any(), any(), any(), eq(herder), any())).thenReturn(true);
 
         herder.tick();
+
+        // now handle the task restart
+        doNothing().when(member).ensureActive();
+        doNothing().when(worker).stopAndAwaitTask(TASK0);
         FutureCallback<Void> callback = new FutureCallback<>();
         herder.restartTask(TASK0, callback);
         herder.tick();
         callback.get(1000L, TimeUnit.MILLISECONDS);
 
-        PowerMock.verifyAll();
+        verify(worker, times(2)).startSourceTask(eq(TASK0), any(), any(), any(), eq(herder), any());
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testRestartUnknownTask() throws Exception {
         // get the initial assignment
-        EasyMock.expect(member.memberId()).andStubReturn("member");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+        when(member.memberId()).thenReturn("member");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
         expectRebalance(1, Collections.emptyList(), Collections.emptyList());
         expectConfigRefreshAndSnapshot(SNAPSHOT);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
-
-        member.wakeup();
-        PowerMock.expectLastCall();
-        member.ensureActive();
-        PowerMock.expectLastCall();
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
+        doNothing().when(member).poll(anyLong());
 
-        PowerMock.replayAll();
+        herder.tick();
 
+        doNothing().when(member).ensureActive();
         FutureCallback<Void> callback = new FutureCallback<>();
-        herder.tick();
         herder.restartTask(new ConnectorTaskId("blah", 0), callback);
         herder.tick();
 
-        try {
-            callback.get(1000L, TimeUnit.MILLISECONDS);
-            fail("Expected NotLeaderException to be raised");
-        } catch (ExecutionException e) {
-            assertTrue(e.getCause() instanceof NotFoundException);
-        }
+        ExecutionException e = assertThrows(ExecutionException.class, () -> callback.get(1000, TimeUnit.MILLISECONDS));
+        assertEquals(NotFoundException.class, e.getCause().getClass());
 
-        PowerMock.verifyAll();
-    }
-
-    @Test
-    public void testRequestProcessingOrder() {
-        final DistributedHerder.DistributedHerderRequest req1 = herder.addRequest(100, null, null);
-        final DistributedHerder.DistributedHerderRequest req2 = herder.addRequest(10, null, null);
-        final DistributedHerder.DistributedHerderRequest req3 = herder.addRequest(200, null, null);
-        final DistributedHerder.DistributedHerderRequest req4 = herder.addRequest(200, null, null);
-
-        assertEquals(req2, herder.requests.pollFirst()); // lowest delay
-        assertEquals(req1, herder.requests.pollFirst()); // next lowest delay
-        assertEquals(req3, herder.requests.pollFirst()); // same delay as req4, but added first
-        assertEquals(req4, herder.requests.pollFirst());
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testRestartTaskRedirectToLeader() throws Exception {
         // get the initial assignment
-        EasyMock.expect(member.memberId()).andStubReturn("member");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+        when(member.memberId()).thenReturn("member");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
         expectRebalance(1, Collections.emptyList(), Collections.emptyList());
         expectConfigRefreshAndSnapshot(SNAPSHOT);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
-
-        // now handle the task restart
-        member.wakeup();
-        PowerMock.expectLastCall();
-        member.ensureActive();
-        PowerMock.expectLastCall();
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
-
-        PowerMock.replayAll();
+        doNothing().when(member).poll(anyLong());
 
         herder.tick();
+
+        // now handle the task restart
+        doNothing().when(member).ensureActive();
         FutureCallback<Void> callback = new FutureCallback<>();
         herder.restartTask(TASK0, callback);
         herder.tick();
 
-        try {
-            callback.get(1000L, TimeUnit.MILLISECONDS);
-            fail("Expected NotLeaderException to be raised");
-        } catch (ExecutionException e) {
-            assertTrue(e.getCause() instanceof NotLeaderException);
-        }
+        ExecutionException e = assertThrows(ExecutionException.class, () -> callback.get(1000, TimeUnit.MILLISECONDS));
+        assertEquals(NotLeaderException.class, e.getCause().getClass());
 
-        PowerMock.verifyAll();
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
     public void testRestartTaskRedirectToOwner() throws Exception {
         // get the initial assignment
-        EasyMock.expect(member.memberId()).andStubReturn("leader");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+        when(member.memberId()).thenReturn("leader");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
         expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true);
         expectConfigRefreshAndSnapshot(SNAPSHOT);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
+
+        when(statusBackingStore.connectors()).thenReturn(Collections.emptySet());
+        doNothing().when(member).poll(anyLong());
+
+        herder.tick();
 
         // now handle the task restart
         String ownerUrl = "ownerUrl";
-        EasyMock.expect(member.ownerUrl(TASK0)).andReturn(ownerUrl);
-        member.wakeup();
-        PowerMock.expectLastCall();
-        member.ensureActive();
-        PowerMock.expectLastCall();
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
+        when(member.ownerUrl(TASK0)).thenReturn(ownerUrl);
+        doNothing().when(member).ensureActive();
 
-        PowerMock.replayAll();
-
-        herder.tick();
         FutureCallback<Void> callback = new FutureCallback<>();
         herder.restartTask(TASK0, callback);
         herder.tick();
 
-        try {
-            callback.get(1000L, TimeUnit.MILLISECONDS);
-            fail("Expected NotLeaderException to be raised");
-        } catch (ExecutionException e) {
-            assertTrue(e.getCause() instanceof NotAssignedException);
-            NotAssignedException notAssignedException = (NotAssignedException) e.getCause();
-            assertEquals(ownerUrl, notAssignedException.forwardUrl());
-        }
+        ExecutionException e = assertThrows(ExecutionException.class, () -> callback.get(1000, TimeUnit.MILLISECONDS));
+        assertEquals(NotAssignedException.class, e.getCause().getClass());
+        assertEquals(ownerUrl, ((NotAssignedException) e.getCause()).forwardUrl());
 
-        PowerMock.verifyAll();
+        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
     }
 
     @Test
-    public void testConnectorConfigAdded() {
+    public void testRequestProcessingOrder() {
+        final DistributedHerder.DistributedHerderRequest req1 = herder.addRequest(100, null, null);
+        final DistributedHerder.DistributedHerderRequest req2 = herder.addRequest(10, null, null);
+        final DistributedHerder.DistributedHerderRequest req3 = herder.addRequest(200, null, null);
+        final DistributedHerder.DistributedHerderRequest req4 = herder.addRequest(200, null, null);
+
+        assertEquals(req2, herder.requests.pollFirst()); // lowest delay
+        assertEquals(req1, herder.requests.pollFirst()); // next lowest delay
+        assertEquals(req3, herder.requests.pollFirst()); // same delay as req4, but added first
+        assertEquals(req4, herder.requests.pollFirst());
+    }
+
+    @Test
+    public void testConnectorConfigAdded() throws Exception {
         // If a connector was added, we need to rebalance
-        EasyMock.expect(member.memberId()).andStubReturn("member");
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+        when(member.memberId()).thenReturn("member");
+        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
 
         // join, no configs so no need to catch up on config topic
         expectRebalance(-1, Collections.emptyList(), Collections.emptyList());
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
+        doNothing().when(member).poll(anyLong());
+
+        herder.tick(); // join
 
-        // apply config
-        member.wakeup();
-        PowerMock.expectLastCall();
-        member.ensureActive();
-        PowerMock.expectLastCall();
         // Checks for config updates and starts rebalance
-        EasyMock.expect(configBackingStore.snapshot()).andReturn(SNAPSHOT);
-        member.requestRejoin();
-        PowerMock.expectLastCall();
+        doNothing().when(configBackingStore).refresh(anyLong(), any());
+        when(configBackingStore.snapshot()).thenReturn(SNAPSHOT);
         // Performs rebalance and gets new assignment
         expectRebalance(Collections.emptyList(), Collections.emptyList(),
                 ConnectProtocol.Assignment.NO_ERROR, 1, Arrays.asList(CONN1), Collections.emptyList());
-        EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
-        Capture<Callback<TargetState>> onStart = newCapture();
-        worker.startConnector(EasyMock.eq(CONN1), EasyMock.anyObject(), EasyMock.anyObject(),
-                EasyMock.eq(herder), EasyMock.eq(TargetState.STARTED), capture(onStart));
-        PowerMock.expectLastCall().andAnswer(() -> {
+
+        ArgumentCaptor<Callback<TargetState>> onStart = ArgumentCaptor.forClass(Callback.class);
+        doAnswer(invocation -> {
             onStart.getValue().onCompletion(null, TargetState.STARTED);
             return true;
-        });
-        member.wakeup();
-        PowerMock.expectLastCall();
-        expectExecuteTaskReconfiguration(true, conn1SinkConfig, () -> TASK_CONFIGS);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
+        }).when(worker).startConnector(eq(CONN1), any(), any(), eq(herder), eq(TargetState.STARTED), onStart.capture());
+        expectExecuteTaskReconfiguration(true, conn1SinkConfig, invocation -> TASK_CONFIGS);
 
-        PowerMock.replayAll();
+        herder.tick(); // apply config
 
-        herder.tick(); // join
+        doNothing().when(member).ensureActive();
         configUpdateListener.onConnectorConfigUpdate(CONN1); // read updated config

Review Comment:
   Yikes, the semantics of the test were changed completely here, good catch! Your understanding of the original test looks correct to me and I've updated the test to match the original semantics.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org