DistributedHerder extends AbstractHerder implements Runnable { @Override public void putConnectorConfig(final String connName, final Map<String, String> config, final boolean allowReplace, final Callback<Created<ConnectorInfo>> callback) { log.trace("Submitting connector config write request {}", connName); addRequest( new Callable<Void>() { @Override public Void call() throws Exception { log.trace("Handling connector config request {}", connName); if (!isLeader()) { callback.onCompletion(new NotLeaderException("Only the leader can set connector configs.", leaderUrl()), null); return null; } boolean exists = configState.contains(connName); if (!allowReplace && exists) { callback.onCompletion(new AlreadyExistsException("Connector " + connName + " already exists"), null); return null; } if (config == null) { if (!exists) { callback.onCompletion(new NotFoundException("Connector " + connName + " not found"), null); } else { log.trace("Removing connector config {} {} {}", connName, allowReplace, configState.connectors()); configBackingStore.removeConnectorConfig(connName); callback.onCompletion(null, new Created<ConnectorInfo>(false, null)); } return null; } log.trace("Submitting connector config {} {} {}", connName, allowReplace, configState.connectors()); configBackingStore.putConnectorConfig(connName, config); ConnectorInfo info = new ConnectorInfo(connName, config, configState.tasks(connName)); callback.onCompletion(null, new Created<>(!exists, info)); 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 testPutConnectorConfig() throws Exception { EasyMock.expect(member.memberId()).andStubReturn("leader"); expectRebalance(1, Arrays.asList(CONN1), Collections.<ConnectorTaskId>emptyList()); expectPostRebalanceCatchup(SNAPSHOT); worker.startConnector(EasyMock.<ConnectorConfig>anyObject(), EasyMock.<ConnectorContext>anyObject(), EasyMock.eq(herder), EasyMock.eq(TargetState.STARTED)); PowerMock.expectLastCall(); EasyMock.expect(worker.isRunning(CONN1)).andReturn(true); EasyMock.expect(worker.connectorTaskConfigs(CONN1, MAX_TASKS, null)).andReturn(TASK_CONFIGS); member.wakeup(); PowerMock.expectLastCall().anyTimes(); member.poll(EasyMock.anyInt()); PowerMock.expectLastCall(); member.ensureActive(); PowerMock.expectLastCall(); configStorage.putConnectorConfig(CONN1, CONN1_CONFIG_UPDATED); PowerMock.expectLastCall().andAnswer(new IAnswer<Object>() { @Override public Object answer() throws Throwable { configUpdateListener.onConnectorConfigUpdate(CONN1); return null; } }); EasyMock.expect(configStorage.snapshot()).andReturn(SNAPSHOT_UPDATED_CONN1_CONFIG); worker.stopConnector(CONN1); PowerMock.expectLastCall(); Capture<ConnectorConfig> capturedUpdatedConfig = EasyMock.newCapture(); worker.startConnector(EasyMock.capture(capturedUpdatedConfig), EasyMock.<ConnectorContext>anyObject(), EasyMock.eq(herder), EasyMock.eq(TargetState.STARTED)); PowerMock.expectLastCall(); EasyMock.expect(worker.isRunning(CONN1)).andReturn(true); EasyMock.expect(worker.connectorTaskConfigs(CONN1, MAX_TASKS, null)).andReturn(TASK_CONFIGS); member.poll(EasyMock.anyInt()); PowerMock.expectLastCall(); member.ensureActive(); PowerMock.expectLastCall(); member.poll(EasyMock.anyInt()); PowerMock.expectLastCall(); PowerMock.replayAll(); FutureCallback<Map<String, String>> connectorConfigCb = new FutureCallback<>(); herder.connectorConfig(CONN1, connectorConfigCb); herder.tick(); assertTrue(connectorConfigCb.isDone()); assertEquals(CONN1_CONFIG, connectorConfigCb.get()); FutureCallback<Herder.Created<ConnectorInfo>> putConfigCb = new FutureCallback<>(); herder.putConnectorConfig(CONN1, CONN1_CONFIG_UPDATED, true, putConfigCb); herder.tick(); assertTrue(putConfigCb.isDone()); ConnectorInfo updatedInfo = new ConnectorInfo(CONN1, CONN1_CONFIG_UPDATED, Arrays.asList(TASK0, TASK1, TASK2)); assertEquals(new Herder.Created<>(false, updatedInfo), putConfigCb.get()); connectorConfigCb = new FutureCallback<>(); herder.connectorConfig(CONN1, connectorConfigCb); herder.tick(); assertTrue(connectorConfigCb.isDone()); assertEquals(CONN1_CONFIG_UPDATED, connectorConfigCb.get()); PowerMock.verifyAll(); }