@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();
}