@Test public void testRestartTask() throws Exception { ConnectorTaskId taskId = new ConnectorTaskId(CONNECTOR_NAME, 0); expectAdd(SourceSink.SOURCE); Map<String, String> connectorConfig = connectorConfig(SourceSink.SOURCE); expectConfigValidation(connectorConfig); worker.stopAndAwaitTask(taskId); EasyMock.expectLastCall(); worker.startTask(taskId, connectorConfig, taskConfig(SourceSink.SOURCE), herder, TargetState.STARTED); EasyMock.expectLastCall().andReturn(true); PowerMock.replayAll(); herder.putConnectorConfig(CONNECTOR_NAME, connectorConfig, false, createCallback); FutureCallback<Void> cb = new FutureCallback<>(); herder.restartTask(taskId, cb); cb.get(1000L, TimeUnit.MILLISECONDS); PowerMock.verifyAll(); }
@Override public synchronized void restartTask(ConnectorTaskId taskId, Callback<Void> cb) { if (!configState.contains(taskId.connector())) cb.onCompletion(new NotFoundException("Connector " + taskId.connector() + " not found", null), null); Map<String, String> taskConfigProps = configState.taskConfig(taskId); if (taskConfigProps == null) cb.onCompletion(new NotFoundException("Task " + taskId + " not found", null), null); Map<String, String> connConfigProps = configState.connectorConfig(taskId.connector()); TargetState targetState = configState.targetState(taskId.connector()); worker.stopAndAwaitTask(taskId); if (worker.startTask(taskId, connConfigProps, taskConfigProps, this, targetState)) cb.onCompletion(null, null); else cb.onCompletion(new ConnectException("Failed to start task: " + taskId), null); }
StandaloneHerder extends AbstractHerder { @Override public synchronized void restartTask(ConnectorTaskId taskId, Callback<Void> cb) { if (!configState.contains(taskId.connector())) cb.onCompletion(new NotFoundException("Connector " + taskId.connector() + " not found", null), null); Map<String, String> taskConfigProps = configState.taskConfig(taskId); if (taskConfigProps == null) cb.onCompletion(new NotFoundException("Task " + taskId + " not found", null), null); Map<String, String> connConfigProps = configState.connectorConfig(taskId.connector()); TargetState targetState = configState.targetState(taskId.connector()); worker.stopAndAwaitTask(taskId); if (worker.startTask(taskId, connConfigProps, taskConfigProps, this, targetState)) cb.onCompletion(null, null); else cb.onCompletion(new ConnectException("Failed to start task: " + taskId), null); } }
StandaloneHerder extends AbstractHerder { @Override public synchronized void restartTask(ConnectorTaskId taskId, Callback<Void> cb) { if (!configState.contains(taskId.connector())) cb.onCompletion(new NotFoundException("Connector " + taskId.connector() + " not found", null), null); Map<String, String> taskConfigProps = configState.taskConfig(taskId); if (taskConfigProps == null) cb.onCompletion(new NotFoundException("Task " + taskId + " not found", null), null); Map<String, String> connConfigProps = configState.connectorConfig(taskId.connector()); TargetState targetState = configState.targetState(taskId.connector()); worker.stopAndAwaitTask(taskId); if (worker.startTask(taskId, connConfigProps, taskConfigProps, this, targetState)) cb.onCompletion(null, null); else cb.onCompletion(new ConnectException("Failed to start task: " + taskId), null); } StandaloneHerder(Worker worker);   StandaloneHerder(Worker worker,
                     String workerId,
                     StatusBackingStore statusBackingStore,
                     MemoryConfigBackingStore configBackingStore); }
StandaloneHerder extends AbstractHerder { @Override public synchronized void restartTask(ConnectorTaskId taskId, Callback<Void> cb) { if (!configState.contains(taskId.connector())) cb.onCompletion(new NotFoundException("Connector " + taskId.connector() + " not found", null), null); Map<String, String> taskConfigProps = configState.taskConfig(taskId); if (taskConfigProps == null) cb.onCompletion(new NotFoundException("Task " + taskId + " not found", null), null); Map<String, String> connConfigProps = configState.connectorConfig(taskId.connector()); TargetState targetState = configState.targetState(taskId.connector()); worker.stopAndAwaitTask(taskId); if (worker.startTask(taskId, connConfigProps, taskConfigProps, this, targetState)) cb.onCompletion(null, null); else cb.onCompletion(new ConnectException("Failed to start task: " + taskId), null); } StandaloneHerder(Worker worker);   StandaloneHerder(Worker worker,
                     String workerId,
                     StatusBackingStore statusBackingStore,
                     MemoryConfigBackingStore configBackingStore); synchronized void start(); synchronized void stop(); @Override int generation(); @Override synchronized void connectors(Callback<Collection<String>> callback); @Override synchronized void connectorInfo(String connName, Callback<ConnectorInfo> callback); @Override void connectorConfig(String connName, final Callback<Map<String, String>> callback); @Override synchronized void deleteConnectorConfig(String connName, Callback<Created<ConnectorInfo>> callback); @Override synchronized void putConnectorConfig(String connName,
                                                final Map<String, String> config,
                                                boolean allowReplace,
                                                final Callback<Created<ConnectorInfo>> callback); @Override synchronized void requestTaskReconfiguration(String connName); @Override synchronized void taskConfigs(String connName, Callback<List<TaskInfo>> callback); @Override void putTaskConfigs(String connName, List<Map<String, String>> configs, Callback<Void> callback); @Override synchronized void restartTask(ConnectorTaskId taskId, Callback<Void> cb); @Override synchronized void restartConnector(String connName, Callback<Void> cb); }
StandaloneHerder extends AbstractHerder { @Override public synchronized void restartTask(ConnectorTaskId taskId, Callback<Void> cb) { if (!configState.contains(taskId.connector())) cb.onCompletion(new NotFoundException("Connector " + taskId.connector() + " not found", null), null); Map<String, String> taskConfigProps = configState.taskConfig(taskId); if (taskConfigProps == null) cb.onCompletion(new NotFoundException("Task " + taskId + " not found", null), null); Map<String, String> connConfigProps = configState.connectorConfig(taskId.connector()); TargetState targetState = configState.targetState(taskId.connector()); worker.stopAndAwaitTask(taskId); if (worker.startTask(taskId, connConfigProps, taskConfigProps, this, targetState)) cb.onCompletion(null, null); else cb.onCompletion(new ConnectException("Failed to start task: " + taskId), null); } StandaloneHerder(Worker worker);   StandaloneHerder(Worker worker,
                     String workerId,
                     StatusBackingStore statusBackingStore,
                     MemoryConfigBackingStore configBackingStore); synchronized void start(); synchronized void stop(); @Override int generation(); @Override synchronized void connectors(Callback<Collection<String>> callback); @Override synchronized void connectorInfo(String connName, Callback<ConnectorInfo> callback); @Override void connectorConfig(String connName, final Callback<Map<String, String>> callback); @Override synchronized void deleteConnectorConfig(String connName, Callback<Created<ConnectorInfo>> callback); @Override synchronized void putConnectorConfig(String connName,
                                                final Map<String, String> config,
                                                boolean allowReplace,
                                                final Callback<Created<ConnectorInfo>> callback); @Override synchronized void requestTaskReconfiguration(String connName); @Override synchronized void taskConfigs(String connName, Callback<List<TaskInfo>> callback); @Override void putTaskConfigs(String connName, List<Map<String, String>> configs, Callback<Void> callback); @Override synchronized void restartTask(ConnectorTaskId taskId, Callback<Void> cb); @Override synchronized void restartConnector(String connName, Callback<Void> cb);  }