@Test public void testPutConnectorConfig() throws Exception { expectConfigure(); expectStart(Collections.EMPTY_LIST, Collections.EMPTY_MAP); expectConvertWriteAndRead( CONNECTOR_CONFIG_KEYS.get(0), KafkaConfigBackingStore.CONNECTOR_CONFIGURATION_V0, CONFIGS_SERIALIZED.get(0), "properties", SAMPLE_CONFIGS.get(0)); configUpdateListener.onConnectorConfigUpdate(CONNECTOR_IDS.get(0)); EasyMock.expectLastCall(); expectConvertWriteAndRead( CONNECTOR_CONFIG_KEYS.get(1), KafkaConfigBackingStore.CONNECTOR_CONFIGURATION_V0, CONFIGS_SERIALIZED.get(1), "properties", SAMPLE_CONFIGS.get(1)); configUpdateListener.onConnectorConfigUpdate(CONNECTOR_IDS.get(1)); EasyMock.expectLastCall(); expectConnectorRemoval(CONNECTOR_CONFIG_KEYS.get(1), TARGET_STATE_KEYS.get(1)); configUpdateListener.onConnectorConfigRemove(CONNECTOR_IDS.get(1)); EasyMock.expectLastCall(); expectStop(); PowerMock.replayAll(); configStorage.setupAndCreateKafkaBasedLog(TOPIC, DEFAULT_DISTRIBUTED_CONFIG); configStorage.start(); ClusterConfigState configState = configStorage.snapshot(); assertEquals(-1, configState.offset()); assertNull(configState.connectorConfig(CONNECTOR_IDS.get(0))); assertNull(configState.connectorConfig(CONNECTOR_IDS.get(1))); configStorage.putConnectorConfig(CONNECTOR_IDS.get(0), SAMPLE_CONFIGS.get(0)); configState = configStorage.snapshot(); assertEquals(1, configState.offset()); assertEquals(SAMPLE_CONFIGS.get(0), configState.connectorConfig(CONNECTOR_IDS.get(0))); assertNull(configState.connectorConfig(CONNECTOR_IDS.get(1))); configStorage.putConnectorConfig(CONNECTOR_IDS.get(1), SAMPLE_CONFIGS.get(1)); configState = configStorage.snapshot(); assertEquals(2, configState.offset()); assertEquals(SAMPLE_CONFIGS.get(0), configState.connectorConfig(CONNECTOR_IDS.get(0))); assertEquals(SAMPLE_CONFIGS.get(1), configState.connectorConfig(CONNECTOR_IDS.get(1))); configStorage.removeConnectorConfig(CONNECTOR_IDS.get(1)); configState = configStorage.snapshot(); assertEquals(4, configState.offset()); assertEquals(SAMPLE_CONFIGS.get(0), configState.connectorConfig(CONNECTOR_IDS.get(0))); assertNull(configState.connectorConfig(CONNECTOR_IDS.get(1))); assertNull(configState.targetState(CONNECTOR_IDS.get(1))); configStorage.stop(); PowerMock.verifyAll(); }
@Override public void putConnectorConfig(String connector, Map<String, String> properties) { log.debug("Writing connector configuration {} for connector {} configuration", properties, connector); Struct connectConfig = new Struct(CONNECTOR_CONFIGURATION_V0); connectConfig.put("properties", properties); byte[] serializedConfig = converter.fromConnectData(topic, CONNECTOR_CONFIGURATION_V0, connectConfig); updateConnectorConfig(connector, serializedConfig); }
KafkaConfigBackingStore implements ConfigBackingStore { @Override public void putConnectorConfig(String connector, Map<String, String> properties) { log.debug("Writing connector configuration {} for connector {} configuration", properties, connector); Struct connectConfig = new Struct(CONNECTOR_CONFIGURATION_V0); connectConfig.put("properties", properties); byte[] serializedConfig = converter.fromConnectData(topic, CONNECTOR_CONFIGURATION_V0, connectConfig); updateConnectorConfig(connector, serializedConfig); } }
KafkaConfigBackingStore implements ConfigBackingStore { @Override public void putConnectorConfig(String connector, Map<String, String> properties) { log.debug("Writing connector configuration {} for connector {} configuration", properties, connector); Struct connectConfig = new Struct(CONNECTOR_CONFIGURATION_V0); connectConfig.put("properties", properties); byte[] serializedConfig = converter.fromConnectData(topic, CONNECTOR_CONFIGURATION_V0, connectConfig); updateConnectorConfig(connector, serializedConfig); } KafkaConfigBackingStore(Converter converter, WorkerConfig config); }
KafkaConfigBackingStore implements ConfigBackingStore { @Override public void putConnectorConfig(String connector, Map<String, String> properties) { log.debug("Writing connector configuration {} for connector {} configuration", properties, connector); Struct connectConfig = new Struct(CONNECTOR_CONFIGURATION_V0); connectConfig.put("properties", properties); byte[] serializedConfig = converter.fromConnectData(topic, CONNECTOR_CONFIGURATION_V0, connectConfig); updateConnectorConfig(connector, serializedConfig); } KafkaConfigBackingStore(Converter converter, WorkerConfig config); 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 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); }
KafkaConfigBackingStore implements ConfigBackingStore { @Override public void putConnectorConfig(String connector, Map<String, String> properties) { log.debug("Writing connector configuration {} for connector {} configuration", properties, connector); Struct connectConfig = new Struct(CONNECTOR_CONFIGURATION_V0); connectConfig.put("properties", properties); byte[] serializedConfig = converter.fromConnectData(topic, CONNECTOR_CONFIGURATION_V0, connectConfig); updateConnectorConfig(connector, serializedConfig); } KafkaConfigBackingStore(Converter converter, WorkerConfig config); 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 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; }