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