@Test public void testRestartConnector() throws Exception { expectAdd(SourceSink.SOURCE); Map<String, String> config = connectorConfig(SourceSink.SOURCE); expectConfigValidation(config); worker.stopConnector(CONNECTOR_NAME); EasyMock.expectLastCall().andReturn(true); worker.startConnector(EasyMock.eq(CONNECTOR_NAME), EasyMock.eq(config), EasyMock.anyObject(HerderConnectorContext.class), EasyMock.eq(herder), EasyMock.eq(TargetState.STARTED)); EasyMock.expectLastCall().andReturn(true); PowerMock.replayAll(); herder.putConnectorConfig(CONNECTOR_NAME, config, false, createCallback); FutureCallback<Void> cb = new FutureCallback<>(); herder.restartConnector(CONNECTOR_NAME, cb); cb.get(1000L, TimeUnit.MILLISECONDS); PowerMock.verifyAll(); }
@Override public synchronized void restartConnector(String connName, Callback<Void> cb) { if (!configState.contains(connName)) cb.onCompletion(new NotFoundException("Connector " + connName + " not found", null), null); Map<String, String> config = configState.connectorConfig(connName); worker.stopConnector(connName); if (startConnector(config)) cb.onCompletion(null, null); else cb.onCompletion(new ConnectException("Failed to start connector: " + connName), null); }
StandaloneHerder extends AbstractHerder { @Override public synchronized void restartConnector(String connName, Callback<Void> cb) { if (!configState.contains(connName)) cb.onCompletion(new NotFoundException("Connector " + connName + " not found", null), null); Map<String, String> config = configState.connectorConfig(connName); worker.stopConnector(connName); if (startConnector(config)) cb.onCompletion(null, null); else cb.onCompletion(new ConnectException("Failed to start connector: " + connName), null); } }
StandaloneHerder extends AbstractHerder { @Override public synchronized void restartConnector(String connName, Callback<Void> cb) { if (!configState.contains(connName)) cb.onCompletion(new NotFoundException("Connector " + connName + " not found", null), null); Map<String, String> config = configState.connectorConfig(connName); worker.stopConnector(connName); if (startConnector(config)) cb.onCompletion(null, null); else cb.onCompletion(new ConnectException("Failed to start connector: " + connName), null); } StandaloneHerder(Worker worker);   StandaloneHerder(Worker worker,
                     String workerId,
                     StatusBackingStore statusBackingStore,
                     MemoryConfigBackingStore configBackingStore); }
StandaloneHerder extends AbstractHerder { @Override public synchronized void restartConnector(String connName, Callback<Void> cb) { if (!configState.contains(connName)) cb.onCompletion(new NotFoundException("Connector " + connName + " not found", null), null); Map<String, String> config = configState.connectorConfig(connName); worker.stopConnector(connName); if (startConnector(config)) cb.onCompletion(null, null); else cb.onCompletion(new ConnectException("Failed to start connector: " + connName), 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 restartConnector(String connName, Callback<Void> cb) { if (!configState.contains(connName)) cb.onCompletion(new NotFoundException("Connector " + connName + " not found", null), null); Map<String, String> config = configState.connectorConfig(connName); worker.stopConnector(connName); if (startConnector(config)) cb.onCompletion(null, null); else cb.onCompletion(new ConnectException("Failed to start connector: " + connName), 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);  }