KafkaConfigBackingStore implements ConfigBackingStore { @Override public void putTaskConfigs(String connector, List<Map<String, String>> configs) { try { configLog.readToEnd().get(READ_TO_END_TIMEOUT_MS, TimeUnit.MILLISECONDS); } catch (InterruptedException | ExecutionException | TimeoutException e) { log.error("Failed to write root configuration to Kafka: ", e); throw new ConnectException("Error writing root configuration to Kafka", e); } int taskCount = configs.size(); int index = 0; for (Map<String, String> taskConfig: configs) { Struct connectConfig = new Struct(TASK_CONFIGURATION_V0); connectConfig.put("properties", taskConfig); byte[] serializedConfig = converter.fromConnectData(topic, TASK_CONFIGURATION_V0, connectConfig); log.debug("Writing configuration for task " + index + " configuration: " + taskConfig); ConnectorTaskId connectorTaskId = new ConnectorTaskId(connector, index); configLog.send(TASK_KEY(connectorTaskId), serializedConfig); index++; } try { if (taskCount > 0) { configLog.readToEnd().get(READ_TO_END_TIMEOUT_MS, TimeUnit.MILLISECONDS); } Struct connectConfig = new Struct(CONNECTOR_TASKS_COMMIT_V0); connectConfig.put("tasks", taskCount); byte[] serializedConfig = converter.fromConnectData(topic, CONNECTOR_TASKS_COMMIT_V0, connectConfig); log.debug("Writing commit for connector " + connector + " with " + taskCount + " tasks."); configLog.send(COMMIT_TASKS_KEY(connector), serializedConfig); configLog.readToEnd().get(READ_TO_END_TIMEOUT_MS, TimeUnit.MILLISECONDS); } catch (InterruptedException | ExecutionException | TimeoutException e) { log.error("Failed to write root configuration to Kafka: ", e); throw new ConnectException("Error writing root configuration to Kafka", e); } } KafkaConfigBackingStore(Converter converter); static String TARGET_STATE_KEY(String connectorName); static String CONNECTOR_KEY(String connectorName); static String TASK_KEY(ConnectorTaskId taskId); static String COMMIT_TASKS_KEY(String connectorName); @Override void setUpdateListener(UpdateListener listener); @Override void configure(WorkerConfig config); @Override void start(); @Override void stop(); @Override ClusterConfigState snapshot(); @Override boolean contains(String connector); @Override void putConnectorConfig(String connector, Map<String, String> properties); @Override void removeConnectorConfig(String connector); @Override void removeTaskConfigs(String connector); @Override void putTaskConfigs(String connector, List<Map<String, String>> configs); @Override void refresh(long timeout, TimeUnit unit); @Override void putTargetState(String connector, TargetState state); static final String TARGET_STATE_PREFIX; static final String CONNECTOR_PREFIX; static final String TASK_PREFIX; static final String COMMIT_TASKS_PREFIX; static final Schema CONNECTOR_CONFIGURATION_V0; static final Schema TASK_CONFIGURATION_V0; static final Schema CONNECTOR_TASKS_COMMIT_V0; static final Schema TARGET_STATE_V0; }
@Test public void testPutTaskConfigs() throws Exception { expectConfigure(); expectStart(Collections.EMPTY_LIST, Collections.EMPTY_MAP); expectReadToEnd(new LinkedHashMap<String, byte[]>()); expectConvertWriteRead( TASK_CONFIG_KEYS.get(0), KafkaConfigBackingStore.TASK_CONFIGURATION_V0, CONFIGS_SERIALIZED.get(0), "properties", SAMPLE_CONFIGS.get(0)); expectConvertWriteRead( TASK_CONFIG_KEYS.get(1), KafkaConfigBackingStore.TASK_CONFIGURATION_V0, CONFIGS_SERIALIZED.get(1), "properties", SAMPLE_CONFIGS.get(1)); expectReadToEnd(new LinkedHashMap<String, byte[]>()); expectConvertWriteRead( COMMIT_TASKS_CONFIG_KEYS.get(0), KafkaConfigBackingStore.CONNECTOR_TASKS_COMMIT_V0, CONFIGS_SERIALIZED.get(2), "tasks", 2); configUpdateListener.onTaskConfigUpdate(Arrays.asList(TASK_IDS.get(0), TASK_IDS.get(1))); EasyMock.expectLastCall(); LinkedHashMap<String, byte[]> serializedConfigs = new LinkedHashMap<>(); serializedConfigs.put(TASK_CONFIG_KEYS.get(0), CONFIGS_SERIALIZED.get(0)); serializedConfigs.put(TASK_CONFIG_KEYS.get(1), CONFIGS_SERIALIZED.get(1)); serializedConfigs.put(COMMIT_TASKS_CONFIG_KEYS.get(0), CONFIGS_SERIALIZED.get(2)); expectReadToEnd(serializedConfigs); expectStop(); PowerMock.replayAll(); configStorage.configure(DEFAULT_DISTRIBUTED_CONFIG); configStorage.start(); whiteboxAddConnector(CONNECTOR_IDS.get(0), SAMPLE_CONFIGS.get(0), Collections.EMPTY_LIST); ClusterConfigState configState = configStorage.snapshot(); assertEquals(-1, configState.offset()); assertNull(configState.taskConfig(TASK_IDS.get(0))); assertNull(configState.taskConfig(TASK_IDS.get(1))); List<Map<String, String>> taskConfigs = Arrays.asList(SAMPLE_CONFIGS.get(0), SAMPLE_CONFIGS.get(1)); configStorage.putTaskConfigs("connector1", taskConfigs); configState = configStorage.snapshot(); assertEquals(3, configState.offset()); String connectorName = CONNECTOR_IDS.get(0); assertEquals(Arrays.asList(connectorName), new ArrayList<>(configState.connectors())); assertEquals(Arrays.asList(TASK_IDS.get(0), TASK_IDS.get(1)), configState.tasks(connectorName)); assertEquals(SAMPLE_CONFIGS.get(0), configState.taskConfig(TASK_IDS.get(0))); assertEquals(SAMPLE_CONFIGS.get(1), configState.taskConfig(TASK_IDS.get(1))); assertEquals(Collections.EMPTY_SET, configState.inconsistentConnectors()); configStorage.stop(); PowerMock.verifyAll(); }