@Override
    public void putTaskCountRecord(String connector, int taskCount) {
        Struct taskCountRecord = new Struct(TASK_COUNT_RECORD_V0);
        taskCountRecord.put("task-count", taskCount);
        byte[] serializedTaskCountRecord = converter.fromConnectData(topic, TASK_COUNT_RECORD_V0, taskCountRecord);
        log.debug("Writing task count record {} for connector {}", taskCount, connector);
        try {
            Timer timer = time.timer(READ_WRITE_TOTAL_TIMEOUT_MS);
            sendPrivileged(TASK_COUNT_RECORD_KEY(connector), serializedTaskCountRecord, timer);
            configLog.readToEnd().get(timer.remainingMs(), TimeUnit.MILLISECONDS);
        } catch (InterruptedException | ExecutionException | TimeoutException e) {
            log.error("Failed to write task count record with {} tasks for connector {} to Kafka: ", taskCount, connector, e);
            throw new ConnectException("Error writing task count record to Kafka", e);
        }
    }
Write a task count record for a connector to persistent storage and wait until it has been acknowledged and read back by
tailing the Kafka log with a consumer. {@link #claimWritePrivileges()} must be successfully invoked before calling this method
if the worker is configured to use a fencable producer for writes to the config topic.
@param connector name of the connector
@param taskCount number of tasks used by the connector
@throws IllegalStateException if {@link #claimWritePrivileges()} is required, but was not successfully invoked before
this method was called
@throws PrivilegedWriteException if the worker is configured to use a fencable producer for writes to the config topic
and the write fails
connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaConfigBackingStore.java