test_class.declarator: TOPIC = "connect-configs"
test_class.modifier: private static final
test_class.original_string: private static final String TOPIC = "connect-configs";
test_class.type: String
test_class.var_name: TOPIC
test_class.declarator: DEFAULT_CONFIG_STORAGE_PROPS = new HashMap<>()
test_class.modifier: private static final
test_class.original_string: private static final Map<String, String> DEFAULT_CONFIG_STORAGE_PROPS = new HashMap<>();
test_class.type: Map<String, String>
test_class.var_name: DEFAULT_CONFIG_STORAGE_PROPS
test_class.declarator: DEFAULT_DISTRIBUTED_CONFIG
test_class.modifier: private static final
test_class.original_string: private static final DistributedConfig DEFAULT_DISTRIBUTED_CONFIG;
test_class.type: DistributedConfig
test_class.var_name: DEFAULT_DISTRIBUTED_CONFIG
test_class.declarator: CONNECTOR_IDS = Arrays.asList("connector1", "connector2")
test_class.modifier: private static final
test_class.original_string: private static final List<String> CONNECTOR_IDS = Arrays.asList("connector1", "connector2");
test_class.type: List<String>
test_class.var_name: CONNECTOR_IDS
test_class.declarator: CONNECTOR_CONFIG_KEYS = Arrays.asList("connector-connector1", "connector-connector2")
test_class.modifier: private static final
test_class.original_string: private static final List<String> CONNECTOR_CONFIG_KEYS = Arrays.asList("connector-connector1", "connector-connector2");
test_class.type: List<String>
test_class.var_name: CONNECTOR_CONFIG_KEYS
test_class.declarator: COMMIT_TASKS_CONFIG_KEYS = Arrays.asList("commit-connector1", "commit-connector2")
test_class.modifier: private static final
test_class.original_string: private static final List<String> COMMIT_TASKS_CONFIG_KEYS = Arrays.asList("commit-connector1", "commit-connector2");
test_class.type: List<String>
test_class.var_name: COMMIT_TASKS_CONFIG_KEYS
test_class.declarator: TARGET_STATE_KEYS =  Arrays.asList("target-state-connector1", "target-state-connector2")
test_class.modifier: private static final
test_class.original_string: private static final List<String> TARGET_STATE_KEYS =  Arrays.asList("target-state-connector1", "target-state-connector2");
test_class.type: List<String>
test_class.var_name: TARGET_STATE_KEYS
test_class.declarator: TASK_IDS = Arrays.asList(
              new ConnectorTaskId("connector1", 0),
              new ConnectorTaskId("connector1", 1),
              new ConnectorTaskId("connector2", 0)
      )
test_class.modifier: private static final
test_class.original_string: private static final List<ConnectorTaskId> TASK_IDS = Arrays.asList(
              new ConnectorTaskId("connector1", 0),
              new ConnectorTaskId("connector1", 1),
              new ConnectorTaskId("connector2", 0)
      );
test_class.type: List<ConnectorTaskId>
test_class.var_name: TASK_IDS
test_class.declarator: TASK_CONFIG_KEYS = Arrays.asList("task-connector1-0", "task-connector1-1", "task-connector2-0")
test_class.modifier: private static final
test_class.original_string: private static final List<String> TASK_CONFIG_KEYS = Arrays.asList("task-connector1-0", "task-connector1-1", "task-connector2-0");
test_class.type: List<String>
test_class.var_name: TASK_CONFIG_KEYS
test_class.declarator: SAMPLE_CONFIGS = Arrays.asList(
              Collections.singletonMap("config-key-one", "config-value-one"),
              Collections.singletonMap("config-key-two", "config-value-two"),
              Collections.singletonMap("config-key-three", "config-value-three")
      )
test_class.modifier: private static final
test_class.original_string: private static final List<Map<String, String>> SAMPLE_CONFIGS = Arrays.asList(
              Collections.singletonMap("config-key-one", "config-value-one"),
              Collections.singletonMap("config-key-two", "config-value-two"),
              Collections.singletonMap("config-key-three", "config-value-three")
      );
test_class.type: List<Map<String, String>>
test_class.var_name: SAMPLE_CONFIGS
test_class.declarator: CONNECTOR_CONFIG_STRUCTS = Arrays.asList(
              new Struct(KafkaConfigBackingStore.CONNECTOR_CONFIGURATION_V0).put("properties", SAMPLE_CONFIGS.get(0)),
              new Struct(KafkaConfigBackingStore.CONNECTOR_CONFIGURATION_V0).put("properties", SAMPLE_CONFIGS.get(1)),
              new Struct(KafkaConfigBackingStore.CONNECTOR_CONFIGURATION_V0).put("properties", SAMPLE_CONFIGS.get(2))
      )
test_class.modifier: private static final
test_class.original_string: private static final List<Struct> CONNECTOR_CONFIG_STRUCTS = Arrays.asList(
              new Struct(KafkaConfigBackingStore.CONNECTOR_CONFIGURATION_V0).put("properties", SAMPLE_CONFIGS.get(0)),
              new Struct(KafkaConfigBackingStore.CONNECTOR_CONFIGURATION_V0).put("properties", SAMPLE_CONFIGS.get(1)),
              new Struct(KafkaConfigBackingStore.CONNECTOR_CONFIGURATION_V0).put("properties", SAMPLE_CONFIGS.get(2))
      );
test_class.type: List<Struct>
test_class.var_name: CONNECTOR_CONFIG_STRUCTS
test_class.declarator: TASK_CONFIG_STRUCTS = Arrays.asList(
              new Struct(KafkaConfigBackingStore.TASK_CONFIGURATION_V0).put("properties", SAMPLE_CONFIGS.get(0)),
              new Struct(KafkaConfigBackingStore.TASK_CONFIGURATION_V0).put("properties", SAMPLE_CONFIGS.get(1))
      )
test_class.modifier: private static final
test_class.original_string: private static final List<Struct> TASK_CONFIG_STRUCTS = Arrays.asList(
              new Struct(KafkaConfigBackingStore.TASK_CONFIGURATION_V0).put("properties", SAMPLE_CONFIGS.get(0)),
              new Struct(KafkaConfigBackingStore.TASK_CONFIGURATION_V0).put("properties", SAMPLE_CONFIGS.get(1))
      );
test_class.type: List<Struct>
test_class.var_name: TASK_CONFIG_STRUCTS
test_class.declarator: TARGET_STATE_PAUSED = new Struct(KafkaConfigBackingStore.TARGET_STATE_V0).put("state", "PAUSED")
test_class.modifier: private static final
test_class.original_string: private static final Struct TARGET_STATE_PAUSED = new Struct(KafkaConfigBackingStore.TARGET_STATE_V0).put("state", "PAUSED");
test_class.type: Struct
test_class.var_name: TARGET_STATE_PAUSED
test_class.declarator: TASKS_COMMIT_STRUCT_TWO_TASK_CONNECTOR
              = new Struct(KafkaConfigBackingStore.CONNECTOR_TASKS_COMMIT_V0).put("tasks", 2)
test_class.modifier: private static final
test_class.original_string: private static final Struct TASKS_COMMIT_STRUCT_TWO_TASK_CONNECTOR
              = new Struct(KafkaConfigBackingStore.CONNECTOR_TASKS_COMMIT_V0).put("tasks", 2);
test_class.type: Struct
test_class.var_name: TASKS_COMMIT_STRUCT_TWO_TASK_CONNECTOR
test_class.declarator: TASKS_COMMIT_STRUCT_ZERO_TASK_CONNECTOR
              = new Struct(KafkaConfigBackingStore.CONNECTOR_TASKS_COMMIT_V0).put("tasks", 0)
test_class.modifier: private static final
test_class.original_string: private static final Struct TASKS_COMMIT_STRUCT_ZERO_TASK_CONNECTOR
              = new Struct(KafkaConfigBackingStore.CONNECTOR_TASKS_COMMIT_V0).put("tasks", 0);
test_class.type: Struct
test_class.var_name: TASKS_COMMIT_STRUCT_ZERO_TASK_CONNECTOR
test_class.declarator: CONFIGS_SERIALIZED = Arrays.asList(
              "config-bytes-1".getBytes(), "config-bytes-2".getBytes(), "config-bytes-3".getBytes(),
              "config-bytes-4".getBytes(), "config-bytes-5".getBytes(), "config-bytes-6".getBytes(),
              "config-bytes-7".getBytes(), "config-bytes-8".getBytes(), "config-bytes-9".getBytes()
      )
test_class.modifier: private static final
test_class.original_string: private static final List<byte[]> CONFIGS_SERIALIZED = Arrays.asList(
              "config-bytes-1".getBytes(), "config-bytes-2".getBytes(), "config-bytes-3".getBytes(),
              "config-bytes-4".getBytes(), "config-bytes-5".getBytes(), "config-bytes-6".getBytes(),
              "config-bytes-7".getBytes(), "config-bytes-8".getBytes(), "config-bytes-9".getBytes()
      );
test_class.type: List<byte[]>
test_class.var_name: CONFIGS_SERIALIZED
test_class.declarator: converter
test_class.modifier: @Mock
      private
test_class.original_string: @Mock
      private Converter converter;
test_class.type: Converter
test_class.var_name: converter
test_class.declarator: configUpdateListener
test_class.modifier: @Mock
      private
test_class.original_string: @Mock
      private ConfigBackingStore.UpdateListener configUpdateListener;
test_class.type: ConfigBackingStore.UpdateListener
test_class.var_name: configUpdateListener
test_class.declarator: storeLog
test_class.modifier: @Mock
test_class.original_string: @Mock
      KafkaBasedLog<String, byte[]> storeLog;
test_class.type: KafkaBasedLog<String, byte[]>
test_class.var_name: storeLog
test_class.declarator: configStorage
test_class.modifier: private
test_class.original_string: private KafkaConfigBackingStore configStorage;
test_class.type: KafkaConfigBackingStore
test_class.var_name: configStorage
test_class.declarator: capturedTopic = EasyMock.newCapture()
test_class.modifier: private
test_class.original_string: private Capture<String> capturedTopic = EasyMock.newCapture();
test_class.type: Capture<String>
test_class.var_name: capturedTopic
test_class.declarator: capturedProducerProps = EasyMock.newCapture()
test_class.modifier: private
test_class.original_string: private Capture<Map<String, Object>> capturedProducerProps = EasyMock.newCapture();
test_class.type: Capture<Map<String, Object>>
test_class.var_name: capturedProducerProps
test_class.declarator: capturedConsumerProps = EasyMock.newCapture()
test_class.modifier: private
test_class.original_string: private Capture<Map<String, Object>> capturedConsumerProps = EasyMock.newCapture();
test_class.type: Capture<Map<String, Object>>
test_class.var_name: capturedConsumerProps
test_class.declarator: capturedConsumedCallback = EasyMock.newCapture()
test_class.modifier: private
test_class.original_string: private Capture<Callback<ConsumerRecord<String, byte[]>>> capturedConsumedCallback = EasyMock.newCapture();
test_class.type: Capture<Callback<ConsumerRecord<String, byte[]>>>
test_class.var_name: capturedConsumedCallback
test_class.declarator: logOffset = 0
test_class.modifier: private
test_class.original_string: private long logOffset = 0;
test_class.type: long
test_class.var_name: logOffset
test_class.file: connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaConfigBackingStoreTest.java
test_class.identifier: KafkaConfigBackingStoreTest
test_case.body: @Test
      public void testPutTaskConfigs() throws Exception {
          expectConfigure();
          expectStart(Collections.EMPTY_LIST, Collections.EMPTY_MAP);
  
          // Task configs should read to end, write to the log, read to end, write root, then read to end again
          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); // Starts with 0 tasks, after update has 2
          // As soon as root is rewritten, we should see a callback notifying us that we reconfigured some tasks
          configUpdateListener.onTaskConfigUpdate(Arrays.asList(TASK_IDS.get(0), TASK_IDS.get(1)));
          EasyMock.expectLastCall();
  
          // Records to be read by consumer as it reads to the end of the log
          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();
  
          // Bootstrap as if we had already added the connector, but no tasks had been added yet
          whiteboxAddConnector(CONNECTOR_IDS.get(0), SAMPLE_CONFIGS.get(0), Collections.EMPTY_LIST);
  
          // Null before writing
          ClusterConfigState configState = configStorage.snapshot();
          assertEquals(-1, configState.offset());
          assertNull(configState.taskConfig(TASK_IDS.get(0)));
          assertNull(configState.taskConfig(TASK_IDS.get(1)));
  
          // Writing task task configs should block until all the writes have been performed and the root record update
          // has completed
          List<Map<String, String>> taskConfigs = Arrays.asList(SAMPLE_CONFIGS.get(0), SAMPLE_CONFIGS.get(1));
          configStorage.putTaskConfigs("connector1", taskConfigs);
  
          // Validate root config by listing all connectors and tasks
          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();
      }
test_case.class_method_signature: KafkaConfigBackingStoreTest.testPutTaskConfigs()
test_case.full_signature: @Test public void testPutTaskConfigs()
test_case.identifier: testPutTaskConfigs
test_case.invocations: expectConfigure expectStart expectReadToEnd expectConvertWriteRead get get get expectConvertWriteRead get get get expectReadToEnd expectConvertWriteRead get get onTaskConfigUpdate asList get get expectLastCall put get get put get get put get get expectReadToEnd expectStop replayAll configure start whiteboxAddConnector get get snapshot assertEquals offset assertNull taskConfig get assertNull taskConfig get asList get get putTaskConfigs snapshot assertEquals offset get assertEquals asList connectors assertEquals asList get get tasks assertEquals get taskConfig get assertEquals get taskConfig get assertEquals inconsistentConnectors stop verifyAll
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testPutTaskConfigs()
focal_class.declarator: log = LoggerFactory.getLogger(KafkaConfigBackingStore.class)
focal_class.modifier: private static final
focal_class.original_string: private static final Logger log = LoggerFactory.getLogger(KafkaConfigBackingStore.class);
focal_class.type: Logger
focal_class.var_name: log
focal_class.declarator: TARGET_STATE_PREFIX = "target-state-"
focal_class.modifier: public static final
focal_class.original_string: public static final String TARGET_STATE_PREFIX = "target-state-";
focal_class.type: String
focal_class.var_name: TARGET_STATE_PREFIX
focal_class.declarator: CONNECTOR_PREFIX = "connector-"
focal_class.modifier: public static final
focal_class.original_string: public static final String CONNECTOR_PREFIX = "connector-";
focal_class.type: String
focal_class.var_name: CONNECTOR_PREFIX
focal_class.declarator: TASK_PREFIX = "task-"
focal_class.modifier: public static final
focal_class.original_string: public static final String TASK_PREFIX = "task-";
focal_class.type: String
focal_class.var_name: TASK_PREFIX
focal_class.declarator: COMMIT_TASKS_PREFIX = "commit-"
focal_class.modifier: public static final
focal_class.original_string: public static final String COMMIT_TASKS_PREFIX = "commit-";
focal_class.type: String
focal_class.var_name: COMMIT_TASKS_PREFIX
focal_class.declarator: CONNECTOR_CONFIGURATION_V0 = SchemaBuilder.struct()
              .field("properties", SchemaBuilder.map(Schema.STRING_SCHEMA, Schema.OPTIONAL_STRING_SCHEMA).build())
              .build()
focal_class.modifier: public static final
focal_class.original_string: public static final Schema CONNECTOR_CONFIGURATION_V0 = SchemaBuilder.struct()
              .field("properties", SchemaBuilder.map(Schema.STRING_SCHEMA, Schema.OPTIONAL_STRING_SCHEMA).build())
              .build();
focal_class.type: Schema
focal_class.var_name: CONNECTOR_CONFIGURATION_V0
focal_class.declarator: TASK_CONFIGURATION_V0 = CONNECTOR_CONFIGURATION_V0
focal_class.modifier: public static final
focal_class.original_string: public static final Schema TASK_CONFIGURATION_V0 = CONNECTOR_CONFIGURATION_V0;
focal_class.type: Schema
focal_class.var_name: TASK_CONFIGURATION_V0
focal_class.declarator: CONNECTOR_TASKS_COMMIT_V0 = SchemaBuilder.struct()
              .field("tasks", Schema.INT32_SCHEMA)
              .build()
focal_class.modifier: public static final
focal_class.original_string: public static final Schema CONNECTOR_TASKS_COMMIT_V0 = SchemaBuilder.struct()
              .field("tasks", Schema.INT32_SCHEMA)
              .build();
focal_class.type: Schema
focal_class.var_name: CONNECTOR_TASKS_COMMIT_V0
focal_class.declarator: TARGET_STATE_V0 = SchemaBuilder.struct()
              .field("state", Schema.STRING_SCHEMA)
              .build()
focal_class.modifier: public static final
focal_class.original_string: public static final Schema TARGET_STATE_V0 = SchemaBuilder.struct()
              .field("state", Schema.STRING_SCHEMA)
              .build();
focal_class.type: Schema
focal_class.var_name: TARGET_STATE_V0
focal_class.declarator: READ_TO_END_TIMEOUT_MS = 30000
focal_class.modifier: private static final
focal_class.original_string: private static final long READ_TO_END_TIMEOUT_MS = 30000;
focal_class.type: long
focal_class.var_name: READ_TO_END_TIMEOUT_MS
focal_class.declarator: lock
focal_class.modifier: private final
focal_class.original_string: private final Object lock;
focal_class.type: Object
focal_class.var_name: lock
focal_class.declarator: starting
focal_class.modifier: private
focal_class.original_string: private boolean starting;
focal_class.type: boolean
focal_class.var_name: starting
focal_class.declarator: converter
focal_class.modifier: private final
focal_class.original_string: private final Converter converter;
focal_class.type: Converter
focal_class.var_name: converter
focal_class.declarator: updateListener
focal_class.modifier: private
focal_class.original_string: private UpdateListener updateListener;
focal_class.type: UpdateListener
focal_class.var_name: updateListener
focal_class.declarator: topic
focal_class.modifier: private
focal_class.original_string: private String topic;
focal_class.type: String
focal_class.var_name: topic
focal_class.declarator: configLog
focal_class.modifier: private
focal_class.original_string: private KafkaBasedLog<String, byte[]> configLog;
focal_class.type: KafkaBasedLog<String, byte[]>
focal_class.var_name: configLog
focal_class.declarator: connectorTaskCounts = new HashMap<>()
focal_class.modifier: private
focal_class.original_string: private Map<String, Integer> connectorTaskCounts = new HashMap<>();
focal_class.type: Map<String, Integer>
focal_class.var_name: connectorTaskCounts
focal_class.declarator: connectorConfigs = new HashMap<>()
focal_class.modifier: private
focal_class.original_string: private Map<String, Map<String, String>> connectorConfigs = new HashMap<>();
focal_class.type: Map<String, Map<String, String>>
focal_class.var_name: connectorConfigs
focal_class.declarator: taskConfigs = new HashMap<>()
focal_class.modifier: private
focal_class.original_string: private Map<ConnectorTaskId, Map<String, String>> taskConfigs = new HashMap<>();
focal_class.type: Map<ConnectorTaskId, Map<String, String>>
focal_class.var_name: taskConfigs
focal_class.declarator: inconsistent = new HashSet<>()
focal_class.modifier: private
focal_class.original_string: private Set<String> inconsistent = new HashSet<>();
focal_class.type: Set<String>
focal_class.var_name: inconsistent
focal_class.declarator: offset
focal_class.modifier: private volatile
focal_class.original_string: private volatile long offset;
focal_class.type: long
focal_class.var_name: offset
focal_class.declarator: deferredTaskUpdates = new HashMap<>()
focal_class.modifier: private final
focal_class.original_string: private final Map<String, Map<ConnectorTaskId, Map<String, String>>> deferredTaskUpdates = new HashMap<>();
focal_class.type: Map<String, Map<ConnectorTaskId, Map<String, String>>>
focal_class.var_name: deferredTaskUpdates
focal_class.declarator: connectorTargetStates = new HashMap<>()
focal_class.modifier: private final
focal_class.original_string: private final Map<String, TargetState> connectorTargetStates = new HashMap<>();
focal_class.type: Map<String, TargetState>
focal_class.var_name: connectorTargetStates
focal_class.file: connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaConfigBackingStore.java
focal_class.identifier: KafkaConfigBackingStore
focal_class.interfaces: implements ConfigBackingStore
focal_class.class_method_signature: KafkaConfigBackingStore.TARGET_STATE_KEY(String connectorName)
focal_class.full_signature: public static String TARGET_STATE_KEY(String connectorName)
focal_class.identifier: TARGET_STATE_KEY
focal_class.modifiers: public static
focal_class.parameters: (String connectorName)
focal_class.return: String
focal_class.signature: String TARGET_STATE_KEY(String connectorName)
focal_class.class_method_signature: KafkaConfigBackingStore.CONNECTOR_KEY(String connectorName)
focal_class.full_signature: public static String CONNECTOR_KEY(String connectorName)
focal_class.identifier: CONNECTOR_KEY
focal_class.modifiers: public static
focal_class.parameters: (String connectorName)
focal_class.return: String
focal_class.signature: String CONNECTOR_KEY(String connectorName)
focal_class.class_method_signature: KafkaConfigBackingStore.TASK_KEY(ConnectorTaskId taskId)
focal_class.full_signature: public static String TASK_KEY(ConnectorTaskId taskId)
focal_class.identifier: TASK_KEY
focal_class.modifiers: public static
focal_class.parameters: (ConnectorTaskId taskId)
focal_class.return: String
focal_class.signature: String TASK_KEY(ConnectorTaskId taskId)
focal_class.class_method_signature: KafkaConfigBackingStore.COMMIT_TASKS_KEY(String connectorName)
focal_class.full_signature: public static String COMMIT_TASKS_KEY(String connectorName)
focal_class.identifier: COMMIT_TASKS_KEY
focal_class.modifiers: public static
focal_class.parameters: (String connectorName)
focal_class.return: String
focal_class.signature: String COMMIT_TASKS_KEY(String connectorName)
focal_class.class_method_signature: KafkaConfigBackingStore.KafkaConfigBackingStore(Converter converter)
focal_class.full_signature: public  KafkaConfigBackingStore(Converter converter)
focal_class.identifier: KafkaConfigBackingStore
focal_class.modifiers: public
focal_class.parameters: (Converter converter)
focal_class.signature:  KafkaConfigBackingStore(Converter converter)
focal_class.class_method_signature: KafkaConfigBackingStore.setUpdateListener(UpdateListener listener)
focal_class.full_signature: @Override public void setUpdateListener(UpdateListener listener)
focal_class.identifier: setUpdateListener
focal_class.modifiers: @Override public
focal_class.parameters: (UpdateListener listener)
focal_class.return: void
focal_class.signature: void setUpdateListener(UpdateListener listener)
focal_class.class_method_signature: KafkaConfigBackingStore.configure(WorkerConfig config)
focal_class.full_signature: @Override public void configure(WorkerConfig config)
focal_class.identifier: configure
focal_class.modifiers: @Override public
focal_class.parameters: (WorkerConfig config)
focal_class.return: void
focal_class.signature: void configure(WorkerConfig config)
focal_class.class_method_signature: KafkaConfigBackingStore.start()
focal_class.full_signature: @Override public void start()
focal_class.identifier: start
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void start()
focal_class.class_method_signature: KafkaConfigBackingStore.stop()
focal_class.full_signature: @Override public void stop()
focal_class.identifier: stop
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void stop()
focal_class.class_method_signature: KafkaConfigBackingStore.snapshot()
focal_class.full_signature: @Override public ClusterConfigState snapshot()
focal_class.identifier: snapshot
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: ClusterConfigState
focal_class.signature: ClusterConfigState snapshot()
focal_class.class_method_signature: KafkaConfigBackingStore.contains(String connector)
focal_class.full_signature: @Override public boolean contains(String connector)
focal_class.identifier: contains
focal_class.modifiers: @Override public
focal_class.parameters: (String connector)
focal_class.return: boolean
focal_class.signature: boolean contains(String connector)
focal_class.class_method_signature: KafkaConfigBackingStore.putConnectorConfig(String connector, Map<String, String> properties)
focal_class.full_signature: @Override public void putConnectorConfig(String connector, Map<String, String> properties)
focal_class.identifier: putConnectorConfig
focal_class.modifiers: @Override public
focal_class.parameters: (String connector, Map<String, String> properties)
focal_class.return: void
focal_class.signature: void putConnectorConfig(String connector, Map<String, String> properties)
focal_class.class_method_signature: KafkaConfigBackingStore.removeConnectorConfig(String connector)
focal_class.full_signature: @Override public void removeConnectorConfig(String connector)
focal_class.identifier: removeConnectorConfig
focal_class.modifiers: @Override public
focal_class.parameters: (String connector)
focal_class.return: void
focal_class.signature: void removeConnectorConfig(String connector)
focal_class.class_method_signature: KafkaConfigBackingStore.removeTaskConfigs(String connector)
focal_class.full_signature: @Override public void removeTaskConfigs(String connector)
focal_class.identifier: removeTaskConfigs
focal_class.modifiers: @Override public
focal_class.parameters: (String connector)
focal_class.return: void
focal_class.signature: void removeTaskConfigs(String connector)
focal_class.class_method_signature: KafkaConfigBackingStore.updateConnectorConfig(String connector, byte[] serializedConfig)
focal_class.full_signature: private void updateConnectorConfig(String connector, byte[] serializedConfig)
focal_class.identifier: updateConnectorConfig
focal_class.modifiers: private
focal_class.parameters: (String connector, byte[] serializedConfig)
focal_class.return: void
focal_class.signature: void updateConnectorConfig(String connector, byte[] serializedConfig)
focal_class.class_method_signature: KafkaConfigBackingStore.putTaskConfigs(String connector, List<Map<String, String>> configs)
focal_class.full_signature: @Override public void putTaskConfigs(String connector, List<Map<String, String>> configs)
focal_class.identifier: putTaskConfigs
focal_class.modifiers: @Override public
focal_class.parameters: (String connector, List<Map<String, String>> configs)
focal_class.return: void
focal_class.signature: void putTaskConfigs(String connector, List<Map<String, String>> configs)
focal_class.class_method_signature: KafkaConfigBackingStore.refresh(long timeout, TimeUnit unit)
focal_class.full_signature: @Override public void refresh(long timeout, TimeUnit unit)
focal_class.identifier: refresh
focal_class.modifiers: @Override public
focal_class.parameters: (long timeout, TimeUnit unit)
focal_class.return: void
focal_class.signature: void refresh(long timeout, TimeUnit unit)
focal_class.class_method_signature: KafkaConfigBackingStore.putTargetState(String connector, TargetState state)
focal_class.full_signature: @Override public void putTargetState(String connector, TargetState state)
focal_class.identifier: putTargetState
focal_class.modifiers: @Override public
focal_class.parameters: (String connector, TargetState state)
focal_class.return: void
focal_class.signature: void putTargetState(String connector, TargetState state)
focal_class.class_method_signature: KafkaConfigBackingStore.createKafkaBasedLog(String topic, Map<String, Object> producerProps,
                                                                Map<String, Object> consumerProps, Callback<ConsumerRecord<String, byte[]>> consumedCallback)
focal_class.full_signature: private KafkaBasedLog<String, byte[]> createKafkaBasedLog(String topic, Map<String, Object> producerProps,
                                                                Map<String, Object> consumerProps, Callback<ConsumerRecord<String, byte[]>> consumedCallback)
focal_class.identifier: createKafkaBasedLog
focal_class.modifiers: private
focal_class.parameters: (String topic, Map<String, Object> producerProps,
                                                                Map<String, Object> consumerProps, Callback<ConsumerRecord<String, byte[]>> consumedCallback)
focal_class.return: KafkaBasedLog<String, byte[]>
focal_class.signature: KafkaBasedLog<String, byte[]> createKafkaBasedLog(String topic, Map<String, Object> producerProps,
                                                                Map<String, Object> consumerProps, Callback<ConsumerRecord<String, byte[]>> consumedCallback)
focal_class.class_method_signature: KafkaConfigBackingStore.parseTaskId(String key)
focal_class.full_signature: private ConnectorTaskId parseTaskId(String key)
focal_class.identifier: parseTaskId
focal_class.modifiers: private
focal_class.parameters: (String key)
focal_class.return: ConnectorTaskId
focal_class.signature: ConnectorTaskId parseTaskId(String key)
focal_class.class_method_signature: KafkaConfigBackingStore.taskIds(String connector, Map<ConnectorTaskId, Map<String, String>> configs)
focal_class.full_signature: private Set<Integer> taskIds(String connector, Map<ConnectorTaskId, Map<String, String>> configs)
focal_class.identifier: taskIds
focal_class.modifiers: private
focal_class.parameters: (String connector, Map<ConnectorTaskId, Map<String, String>> configs)
focal_class.return: Set<Integer>
focal_class.signature: Set<Integer> taskIds(String connector, Map<ConnectorTaskId, Map<String, String>> configs)
focal_class.class_method_signature: KafkaConfigBackingStore.completeTaskIdSet(Set<Integer> idSet, int expectedSize)
focal_class.full_signature: private boolean completeTaskIdSet(Set<Integer> idSet, int expectedSize)
focal_class.identifier: completeTaskIdSet
focal_class.modifiers: private
focal_class.parameters: (Set<Integer> idSet, int expectedSize)
focal_class.return: boolean
focal_class.signature: boolean completeTaskIdSet(Set<Integer> idSet, int expectedSize)
focal_class.class_method_signature: KafkaConfigBackingStore.intValue(Object value)
focal_class.full_signature: private static int intValue(Object value)
focal_class.identifier: intValue
focal_class.modifiers: private static
focal_class.parameters: (Object value)
focal_class.return: int
focal_class.signature: int intValue(Object value)
focal_method.body: @Override
      public void putTaskConfigs(String connector, List<Map<String, String>> configs) {
          // Make sure we're at the end of the log. We should be the only writer, but we want to make sure we don't have
          // any outstanding lagging data to consume.
          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();
  
          // Start sending all the individual updates
          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++;
          }
  
          // Finally, send the commit to update the number of tasks and apply the new configs, then wait until we read to
          // the end of the log
          try {
              // Read to end to ensure all the task configs have been written
              if (taskCount > 0) {
                  configLog.readToEnd().get(READ_TO_END_TIMEOUT_MS, TimeUnit.MILLISECONDS);
              }
              // Write the commit message
              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);
  
              // Read to end to ensure all the commit messages have been written
              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);
          }
      }
focal_method.class_method_signature: KafkaConfigBackingStore.putTaskConfigs(String connector, List<Map<String, String>> configs)
focal_method.full_signature: @Override public void putTaskConfigs(String connector, List<Map<String, String>> configs)
focal_method.identifier: putTaskConfigs
focal_method.invocations: get readToEnd error size put fromConnectData debug send TASK_KEY get readToEnd put fromConnectData debug send COMMIT_TASKS_KEY get readToEnd error
focal_method.modifiers: @Override public
focal_method.parameters: (String connector, List<Map<String, String>> configs)
focal_method.return: void
focal_method.signature: void putTaskConfigs(String connector, List<Map<String, String>> configs)