@Override
    public void putTaskConfigs(String connector, List<Map<String, String>> configs) {
        Timer timer = time.timer(READ_WRITE_TOTAL_TIMEOUT_MS);
        // 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(timer.remainingMs(), TimeUnit.MILLISECONDS);
            timer.update();
        } 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();

        // Send all the individual updates
        int index = 0;
        List<ProducerKeyValue> keyValues = new ArrayList<>();
        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 connector '{}' task {}", connector, index);
            ConnectorTaskId connectorTaskId = new ConnectorTaskId(connector, index);
            keyValues.add(new ProducerKeyValue(TASK_KEY(connectorTaskId), serializedConfig));
            index++;
        }

        try {
            sendPrivileged(keyValues, timer);
        } catch (ExecutionException | InterruptedException | TimeoutException e) {
            log.error("Failed to write task configurations to Kafka", e);
            throw new ConnectException("Error writing task configurations to Kafka", e);
        }

        // 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(timer.remainingMs(), TimeUnit.MILLISECONDS);
                timer.update();
            }
            // 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 '{}' with {} tasks.", connector, taskCount);

            sendPrivileged(COMMIT_TASKS_KEY(connector), serializedConfig, timer);

            // Read to end to ensure all the commit messages have been written
            configLog.readToEnd().get(timer.remainingMs(), 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);
        }
    }
@Test
    public void testPutTaskConfigs() throws Exception {
        configStorage.setupAndCreateKafkaBasedLog(TOPIC, config);
        verifyConfigure();
        configStorage.start();
        verify(configLog).start();

        doAnswer(expectReadToEnd(new LinkedHashMap<>()))
                .doAnswer(expectReadToEnd(new LinkedHashMap<>()))
                .doAnswer(expectReadToEnd(new LinkedHashMap<String, byte[]>() {{
                        put(TASK_CONFIG_KEYS.get(0), CONFIGS_SERIALIZED.get(0));
                        put(TASK_CONFIG_KEYS.get(1), CONFIGS_SERIALIZED.get(1));
                        put(COMMIT_TASKS_CONFIG_KEYS.get(0), CONFIGS_SERIALIZED.get(2));
                    }})
                )
                .when(configLog).readToEnd();

        // Task configs should read to end, write to the log, read to end, write root, then read to end again
        expectConvertWriteRead2(
                TASK_CONFIG_KEYS.get(0), KafkaConfigBackingStore.TASK_CONFIGURATION_V0, CONFIGS_SERIALIZED.get(0),
                new Struct(KafkaConfigBackingStore.TASK_CONFIGURATION_V0).put("properties", SAMPLE_CONFIGS.get(0)));
        expectConvertWriteRead2(
                TASK_CONFIG_KEYS.get(1), KafkaConfigBackingStore.TASK_CONFIGURATION_V0, CONFIGS_SERIALIZED.get(1),
                new Struct(KafkaConfigBackingStore.TASK_CONFIGURATION_V0).put("properties", SAMPLE_CONFIGS.get(1)));
        expectConvertWriteRead2(
                COMMIT_TASKS_CONFIG_KEYS.get(0), KafkaConfigBackingStore.CONNECTOR_TASKS_COMMIT_V0, CONFIGS_SERIALIZED.get(2),
                new Struct(KafkaConfigBackingStore.CONNECTOR_TASKS_COMMIT_V0).put("tasks", 2)); // Starts with 0 tasks, after update has 2
        // Bootstrap as if we had already added the connector, but no tasks had been added yet
        addConnector(CONNECTOR_IDS.get(0), SAMPLE_CONFIGS.get(0), Collections.emptyList());

        // 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 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(Collections.singletonList(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());

        // As soon as root is rewritten, we should see a callback notifying us that we reconfigured some tasks
        verify(configUpdateListener).onTaskConfigUpdate(Arrays.asList(TASK_IDS.get(0), TASK_IDS.get(1)));

        configStorage.stop();
        verify(configLog).stop();
    }