@Test public void testPutConnectorConfig() throws Exception { Map<String, String> connConfig = connectorConfig(SourceSink.SOURCE); Map<String, String> newConnConfig = new HashMap<>(connConfig); newConnConfig.put("foo", "bar"); Callback<Map<String, String>> connectorConfigCb = PowerMock.createMock(Callback.class); Callback<Herder.Created<ConnectorInfo>> putConnectorConfigCb = PowerMock.createMock(Callback.class); connector = PowerMock.createMock(BogusSourceConnector.class); expectAdd(SourceSink.SOURCE); Connector connectorMock = PowerMock.createMock(Connector.class); expectConfigValidation(connectorMock, true, connConfig); connectorConfigCb.onCompletion(null, connConfig); EasyMock.expectLastCall(); worker.stopConnector(CONNECTOR_NAME); EasyMock.expectLastCall().andReturn(true); Capture<Map<String, String>> capturedConfig = EasyMock.newCapture(); worker.startConnector(EasyMock.eq(CONNECTOR_NAME), EasyMock.capture(capturedConfig), EasyMock.<ConnectorContext>anyObject(), EasyMock.eq(herder), EasyMock.eq(TargetState.STARTED)); EasyMock.expectLastCall().andReturn(true); EasyMock.expect(worker.isRunning(CONNECTOR_NAME)).andReturn(true); EasyMock.expect(worker.connectorTaskConfigs(CONNECTOR_NAME, DEFAULT_MAX_TASKS, null)) .andReturn(singletonList(taskConfig(SourceSink.SOURCE))); worker.isSinkConnector(CONNECTOR_NAME); EasyMock.expectLastCall().andReturn(false); ConnectorInfo newConnInfo = new ConnectorInfo(CONNECTOR_NAME, newConnConfig, Arrays.asList(new ConnectorTaskId(CONNECTOR_NAME, 0))); putConnectorConfigCb.onCompletion(null, new Herder.Created<>(false, newConnInfo)); EasyMock.expectLastCall(); expectConfigValidation(connectorMock, false, newConnConfig); connectorConfigCb.onCompletion(null, newConnConfig); EasyMock.expectLastCall(); PowerMock.replayAll(); herder.putConnectorConfig(CONNECTOR_NAME, connConfig, false, createCallback); herder.connectorConfig(CONNECTOR_NAME, connectorConfigCb); herder.putConnectorConfig(CONNECTOR_NAME, newConnConfig, true, putConnectorConfigCb); assertEquals("bar", capturedConfig.getValue().get("foo")); herder.connectorConfig(CONNECTOR_NAME, connectorConfigCb); PowerMock.verifyAll(); }
@Override public synchronized void putConnectorConfig(String connName, final Map<String, String> config, boolean allowReplace, final Callback<Created<ConnectorInfo>> callback) { try { if (maybeAddConfigErrors(validateConnectorConfig(config), callback)) { return; } boolean created = false; if (configState.contains(connName)) { if (!allowReplace) { callback.onCompletion(new AlreadyExistsException("Connector " + connName + " already exists"), null); return; } worker.stopConnector(connName); } else { created = true; } if (!startConnector(config)) { callback.onCompletion(new ConnectException("Failed to start connector: " + connName), null); return; } updateConnectorTasks(connName); callback.onCompletion(null, new Created<>(created, createConnectorInfo(connName))); } catch (ConnectException e) { callback.onCompletion(e, null); } }
StandaloneHerder extends AbstractHerder { @Override public synchronized void putConnectorConfig(String connName, final Map<String, String> config, boolean allowReplace, final Callback<Created<ConnectorInfo>> callback) { try { if (maybeAddConfigErrors(validateConnectorConfig(config), callback)) { return; } boolean created = false; if (configState.contains(connName)) { if (!allowReplace) { callback.onCompletion(new AlreadyExistsException("Connector " + connName + " already exists"), null); return; } worker.stopConnector(connName); } else { created = true; } if (!startConnector(config)) { callback.onCompletion(new ConnectException("Failed to start connector: " + connName), null); return; } updateConnectorTasks(connName); callback.onCompletion(null, new Created<>(created, createConnectorInfo(connName))); } catch (ConnectException e) { callback.onCompletion(e, null); } } }
StandaloneHerder extends AbstractHerder { @Override public synchronized void putConnectorConfig(String connName, final Map<String, String> config, boolean allowReplace, final Callback<Created<ConnectorInfo>> callback) { try { if (maybeAddConfigErrors(validateConnectorConfig(config), callback)) { return; } boolean created = false; if (configState.contains(connName)) { if (!allowReplace) { callback.onCompletion(new AlreadyExistsException("Connector " + connName + " already exists"), null); return; } worker.stopConnector(connName); } else { created = true; } if (!startConnector(config)) { callback.onCompletion(new ConnectException("Failed to start connector: " + connName), null); return; } updateConnectorTasks(connName); callback.onCompletion(null, new Created<>(created, createConnectorInfo(connName))); } catch (ConnectException e) { callback.onCompletion(e, null); } } StandaloneHerder(Worker worker);   StandaloneHerder(Worker worker,
                     String workerId,
                     StatusBackingStore statusBackingStore,
                     MemoryConfigBackingStore configBackingStore); }
StandaloneHerder extends AbstractHerder { @Override public synchronized void putConnectorConfig(String connName, final Map<String, String> config, boolean allowReplace, final Callback<Created<ConnectorInfo>> callback) { try { if (maybeAddConfigErrors(validateConnectorConfig(config), callback)) { return; } boolean created = false; if (configState.contains(connName)) { if (!allowReplace) { callback.onCompletion(new AlreadyExistsException("Connector " + connName + " already exists"), null); return; } worker.stopConnector(connName); } else { created = true; } if (!startConnector(config)) { callback.onCompletion(new ConnectException("Failed to start connector: " + connName), null); return; } updateConnectorTasks(connName); callback.onCompletion(null, new Created<>(created, createConnectorInfo(connName))); } catch (ConnectException e) { callback.onCompletion(e, 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 putConnectorConfig(String connName, final Map<String, String> config, boolean allowReplace, final Callback<Created<ConnectorInfo>> callback) { try { if (maybeAddConfigErrors(validateConnectorConfig(config), callback)) { return; } boolean created = false; if (configState.contains(connName)) { if (!allowReplace) { callback.onCompletion(new AlreadyExistsException("Connector " + connName + " already exists"), null); return; } worker.stopConnector(connName); } else { created = true; } if (!startConnector(config)) { callback.onCompletion(new ConnectException("Failed to start connector: " + connName), null); return; } updateConnectorTasks(connName); callback.onCompletion(null, new Created<>(created, createConnectorInfo(connName))); } catch (ConnectException e) { callback.onCompletion(e, 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);  }