DistributedHerder extends AbstractHerder implements Runnable { @Override public synchronized void restartTask(final ConnectorTaskId id, final Callback<Void> callback) { addRequest(new Callable<Void>() { @Override public Void call() throws Exception { if (checkRebalanceNeeded(callback)) return null; if (!configState.connectors().contains(id.connector())) { callback.onCompletion(new NotFoundException("Unknown connector: " + id.connector()), null); return null; } if (configState.taskConfig(id) == null) { callback.onCompletion(new NotFoundException("Unknown task: " + id), null); return null; } if (worker.ownsTask(id)) { try { worker.stopAndAwaitTask(id); startTask(id); callback.onCompletion(null, null); } catch (Throwable t) { callback.onCompletion(t, null); } } else if (isLeader()) { callback.onCompletion(new NotAssignedException("Cannot restart task since it is not assigned to this member", member.ownerUrl(id)), null); } else { callback.onCompletion(new NotLeaderException("Cannot restart task since it is not assigned to this member", leaderUrl()), null); } return null; } }, forwardErrorCallback(callback)); } DistributedHerder(DistributedConfig config,
                             Time time,
                             Worker worker,
                             StatusBackingStore statusBackingStore,
                             ConfigBackingStore configBackingStore,
                             String restUrl);   DistributedHerder(DistributedConfig config,
                      Worker worker,
                      String workerId,
                      StatusBackingStore statusBackingStore,
                      ConfigBackingStore configStorage,
                      WorkerGroupMember member,
                      String restUrl,
                      Time time); @Override void start(); @Override void run(); void tick(); void halt(); @Override void stop(); @Override synchronized void connectors(final Callback<Collection<String>> callback); @Override synchronized void connectorInfo(final String connName, final Callback<ConnectorInfo> callback); @Override void connectorConfig(String connName, final Callback<Map<String, String>> callback); @Override void putConnectorConfig(final String connName, final Map<String, String> config, final boolean allowReplace,
                                   final Callback<Created<ConnectorInfo>> callback); @Override synchronized void requestTaskReconfiguration(final String connName); @Override synchronized void taskConfigs(final String connName, final Callback<List<TaskInfo>> callback); @Override synchronized void putTaskConfigs(final String connName, final List<Map<String, String>> configs, final Callback<Void> callback); @Override synchronized void restartConnector(final String connName, final Callback<Void> callback); @Override synchronized void restartTask(final ConnectorTaskId id, final Callback<Void> callback); @Override int generation();  }
@Test public void testRestartTask() throws Exception { EasyMock.expect(worker.connectorTaskConfigs(CONN1, MAX_TASKS, null)).andStubReturn(TASK_CONFIGS); EasyMock.expect(member.memberId()).andStubReturn("leader"); expectRebalance(1, Collections.<String>emptyList(), Collections.singletonList(TASK0)); expectPostRebalanceCatchup(SNAPSHOT); member.poll(EasyMock.anyInt()); PowerMock.expectLastCall(); worker.startTask(EasyMock.eq(TASK0), EasyMock.<TaskConfig>anyObject(), EasyMock.eq(herder), EasyMock.eq(TargetState.STARTED)); PowerMock.expectLastCall(); member.wakeup(); PowerMock.expectLastCall(); member.ensureActive(); PowerMock.expectLastCall(); member.poll(EasyMock.anyInt()); PowerMock.expectLastCall(); EasyMock.expect(worker.ownsTask(TASK0)).andReturn(true); worker.stopAndAwaitTask(TASK0); PowerMock.expectLastCall(); worker.startTask(EasyMock.eq(TASK0), EasyMock.<TaskConfig>anyObject(), EasyMock.eq(herder), EasyMock.eq(TargetState.STARTED)); PowerMock.expectLastCall(); PowerMock.replayAll(); herder.tick(); FutureCallback<Void> callback = new FutureCallback<>(); herder.restartTask(TASK0, callback); herder.tick(); callback.get(1000L, TimeUnit.MILLISECONDS); PowerMock.verifyAll(); }