content: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements. See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License. You may obtain a copy of the License at
   *
   *    http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.kafka.connect.storage;
  
  import org.apache.kafka.clients.CommonClientConfigs;
  import org.apache.kafka.clients.admin.NewTopic;
  import org.apache.kafka.clients.consumer.ConsumerConfig;
  import org.apache.kafka.clients.consumer.ConsumerRecord;
  import org.apache.kafka.clients.producer.KafkaProducer;
  import org.apache.kafka.clients.producer.Producer;
  import org.apache.kafka.clients.producer.ProducerConfig;
  import org.apache.kafka.clients.producer.ProducerRecord;
  import org.apache.kafka.clients.producer.RecordMetadata;
  import org.apache.kafka.common.IsolationLevel;
  import org.apache.kafka.common.config.ConfigException;
  import org.apache.kafka.common.errors.UnsupportedVersionException;
  import org.apache.kafka.common.serialization.ByteArrayDeserializer;
  import org.apache.kafka.common.serialization.ByteArraySerializer;
  import org.apache.kafka.common.serialization.StringDeserializer;
  import org.apache.kafka.common.serialization.StringSerializer;
  import org.apache.kafka.common.utils.Time;
  import org.apache.kafka.common.utils.Timer;
  import org.apache.kafka.common.utils.Utils;
  import org.apache.kafka.connect.data.Schema;
  import org.apache.kafka.connect.data.SchemaAndValue;
  import org.apache.kafka.connect.data.SchemaBuilder;
  import org.apache.kafka.connect.data.Struct;
  import org.apache.kafka.connect.errors.ConnectException;
  import org.apache.kafka.connect.errors.DataException;
  import org.apache.kafka.connect.runtime.Herder;
  import org.apache.kafka.connect.runtime.RestartRequest;
  import org.apache.kafka.connect.runtime.SessionKey;
  import org.apache.kafka.connect.runtime.TargetState;
  import org.apache.kafka.connect.runtime.WorkerConfig;
  import org.apache.kafka.connect.runtime.WorkerConfigTransformer;
  import org.apache.kafka.connect.runtime.distributed.ConnectProtocolCompatibility;
  import org.apache.kafka.connect.runtime.distributed.DistributedConfig;
  import org.apache.kafka.connect.util.Callback;
  import org.apache.kafka.connect.util.ConnectUtils;
  import org.apache.kafka.connect.util.ConnectorTaskId;
  import org.apache.kafka.connect.util.KafkaBasedLog;
  import org.apache.kafka.connect.util.TopicAdmin;
  
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.time.Duration;
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.Base64;
  import java.util.Collections;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.List;
  import java.util.Map;
  import java.util.Objects;
  import java.util.Set;
  import java.util.TreeSet;
  import java.util.concurrent.ExecutionException;
  import java.util.concurrent.Future;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.TimeoutException;
  import java.util.function.Supplier;
  
  import javax.crypto.spec.SecretKeySpec;
  
  import static org.apache.kafka.connect.runtime.TargetState.PAUSED;
  import static org.apache.kafka.connect.runtime.TargetState.STOPPED;
  import static org.apache.kafka.connect.util.ConnectUtils.className;
  
  /**
   * <p>
   * Provides persistent storage of Kafka Connect connector configurations in a Kafka topic.
   * </p>
   * <p>
   * This class manages both connector and task configurations, among other various configurations. It tracks seven types
   * of records:
   * <ol>
   *   <li> Connector config: map of string -> string configurations passed to the Connector class, with support for
   *   expanding this format if necessary. (Kafka key: connector-[connector-id]).
   *   These configs are *not* ephemeral. They represent the source of truth. If the entire Connect
   *   cluster goes down, this is all that is really needed to recover.
   *   <li> Task configs: map of string -> string configurations passed to the Task class, with support for expanding
   *   this format if necessary. (Kafka key: task-[connector-id]-[task-id]).
   *   These configs are ephemeral; they are stored here to:
   *     <ul>
   *       <li> disseminate them to all workers while ensuring agreement
   *       <li> to allow faster cluster/worker recovery since the common case of recovery (restoring a connector) will simply
   *       result in the same task configuration as before the failure.
   *     </ul>
   *   <li> Task commit "configs": records indicating that previous task config entries should be committed and all task
   *   configs for a connector can be applied. (Kafka key: commit-[connector-id].
   *   This config has two effects. First, it records the number of tasks the connector is currently
   *   running (and can therefore increase/decrease parallelism). Second, because each task config
   *   is stored separately but they need to be applied together to ensure each partition is assigned
   *   to a single task, this record also indicates that task configs for the specified connector
   *   can be "applied" or "committed".
   *   <li> Connector target states: records indicating the {@link TargetState} for a connector
   *   <li> {@link RestartRequest Restart requests}: records representing requests to restart a connector and / or its
   *   tasks. See <a href="https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=181308623">KIP-745</a> for more
   *   details.
   *   <li> Task count records: an integer value that that tracks the number of task producers (for source connectors) that
   *   will have to be fenced out if a connector is reconfigured before bringing up any tasks with the new set of task
   *   configurations. This is required for exactly-once support for source connectors, see
   *   <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-618%3A+Exactly-Once+Support+for+Source+Connectors">KIP-618</a>
   *   for more details.
   *   <li> Session key records: A {@link SessionKey} generated by the leader of the cluster when
   *   {@link ConnectProtocolCompatibility#SESSIONED} is the Connect protocol being used by the cluster. This session key
   *   is used to verify internal REST requests between workers in the cluster. See
   *   <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-507%3A+Securing+Internal+Connect+REST+Endpoints">KIP-507</a>
   *   for more details.
   * </ol>
   * <p/>
   * <p>
   * This configuration is expected to be stored in a *single partition* and *compacted* topic. Using a single partition
   * ensures we can enforce ordering on messages, allowing Kafka to be used as a write ahead log. Compaction allows
   * us to clean up outdated configurations over time. However, this combination has some important implications for
   * the implementation of this class and the configuration state that it may expose.
   * </p>
   * <p>
   * Connector configurations are independent of all other configs, so they are handled easily. Writing a single record
   * is already atomic, so these can be applied as soon as they are read. One connectors config does not affect any
   * others, and they do not need to coordinate with the connector's task configuration at all.
   * </p>
   * <p>
   * The most obvious implication for task configs is the need for the commit messages. This class doesn't currently make
   * use of Kafka's multi-record transactions and instead uses task commit messages to ensure that readers do not
   * end up using inconsistent configs. For example, consider if a connector wrote configs for its tasks,
   * then was reconfigured and only managed to write updated configs for half its tasks. If task configs
   * were applied immediately you could be using half the old configs and half the new configs. In that condition, some
   * partitions may be double-assigned because the old config and new config may use completely different assignments.
   * Therefore, when reading the log, we must buffer config updates for a connector's tasks and only apply atomically them
   * once a commit message has been read.
   * </p>
   * <p>
   * However, there are also further challenges. This simple buffering approach would work fine as long as the entire log was
   * always available, but we would like to be able to enable compaction so our configuration topic does not grow
   * indefinitely. Compaction may break a normal log because old entries will suddenly go missing. A new worker reading
   * from the beginning of the log in order to build up the full current configuration will see task commits, but some
   * records required for those commits will have been removed because the same keys have subsequently been rewritten.
   * For example, if you have a sequence of record keys [connector-foo-config, task-foo-1-config, task-foo-2-config,
   * commit-foo (2 tasks), task-foo-1-config, commit-foo (1 task)], we can end up with a compacted log containing
   * [connector-foo-config, task-foo-2-config, commit-foo (2 tasks), task-foo-1-config, commit-foo (1 task)]. When read
   * back, the first commit will see an invalid state because the first task-foo-1-config has been cleaned up.
   * </p>
   * <p>
   * Compaction can further complicate things if writing new task configs fails mid-write. Consider a similar scenario
   * as the previous one, but in this case both the first and second update will write 2 task configs. However, the
   * second write fails half of the way through:
   * [connector-foo-config, task-foo-1-config, task-foo-2-config, commit-foo (2 tasks), task-foo-1-config]. Now compaction
   * occurs and we're left with
   * [connector-foo-config, task-foo-2-config, commit-foo (2 tasks), task-foo-1-config]. At the first commit, we don't
   * have a complete set of configs. And because of the failure, there is no second commit. We are left in an inconsistent
   * state with no obvious way to resolve the issue -- we can try to keep on reading, but the failed node may never
   * recover and write the updated config. Meanwhile, other workers may have seen the entire log; they will see the second
   * task-foo-1-config waiting to be applied, but will otherwise think everything is ok -- they have a valid set of task
   * configs for connector "foo".
   * </p>
   * <p>
   * Because we can encounter these inconsistencies and addressing them requires support from the rest of the system
   * (resolving the task configuration inconsistencies requires support from the connector instance to regenerate updated
   * configs), this class exposes not only the current set of configs, but also which connectors have inconsistent data.
   * This allows users of this class (i.e., {@link Herder} implementations) to take action to resolve any inconsistencies. These
   * inconsistencies should be rare (as described above, due to compaction combined with leader failures in the middle
   * of updating task configurations).
   * </p>
   * <p>
   * Note that the expectation is that this config storage system has only a single writer at a time.
   * The caller (Herder) must ensure this is the case. In distributed mode this will require forwarding config change
   * requests to the leader in the cluster (i.e. the worker group coordinated by the Kafka broker).
   * </p>
   * <p>
   * Since processing of the config log occurs in a background thread, callers must take care when using accessors.
   * To simplify handling this correctly, this class only exposes a mechanism to snapshot the current state of the cluster.
   * Updates may continue to be applied (and callbacks invoked) in the background. Callers must take care that they are
   * using a consistent snapshot and only update when it is safe. In particular, if task configs are updated which require
   * synchronization across workers to commit offsets and update the configuration, callbacks and updates during the
   * rebalance must be deferred.
   * </p>
   */
  public final class KafkaConfigBackingStore extends KafkaTopicBasedBackingStore implements ConfigBackingStore {
      private static final Logger log = LoggerFactory.getLogger(KafkaConfigBackingStore.class);
  
      public static final String TARGET_STATE_PREFIX = "target-state-";
  
      public static String TARGET_STATE_KEY(String connectorName) {
          return TARGET_STATE_PREFIX + connectorName;
      }
  
      public static final String CONNECTOR_PREFIX = "connector-";
  
      public static String CONNECTOR_KEY(String connectorName) {
          return CONNECTOR_PREFIX + connectorName;
      }
  
      public static final String TASK_PREFIX = "task-";
  
      public static String TASK_KEY(ConnectorTaskId taskId) {
          return TASK_PREFIX + taskId.connector() + "-" + taskId.task();
      }
  
      public static final String COMMIT_TASKS_PREFIX = "commit-";
  
      public static String COMMIT_TASKS_KEY(String connectorName) {
          return COMMIT_TASKS_PREFIX + connectorName;
      }
  
      public static final String TASK_COUNT_RECORD_PREFIX = "tasks-fencing-";
  
      public static String TASK_COUNT_RECORD_KEY(String connectorName) {
          return TASK_COUNT_RECORD_PREFIX + connectorName;
      }
  
      public static final String SESSION_KEY_KEY = "session-key";
  
      // Note that while using real serialization for values as we have here, but ad hoc string serialization for keys,
      // isn't ideal, we use this approach because it avoids any potential problems with schema evolution or
      // converter/serializer changes causing keys to change. We need to absolutely ensure that the keys remain precisely
      // the same.
      public static final Schema CONNECTOR_CONFIGURATION_V0 = SchemaBuilder.struct()
              .field("properties", SchemaBuilder.map(Schema.STRING_SCHEMA, Schema.OPTIONAL_STRING_SCHEMA).build())
              .build();
      public static final Schema TASK_CONFIGURATION_V0 = CONNECTOR_CONFIGURATION_V0;
      public static final Schema CONNECTOR_TASKS_COMMIT_V0 = SchemaBuilder.struct()
              .field("tasks", Schema.INT32_SCHEMA)
              .build();
      public static final Schema TARGET_STATE_V0 = SchemaBuilder.struct()
              .field("state", Schema.STRING_SCHEMA)
              .build();
      public static final Schema TARGET_STATE_V1 = SchemaBuilder.struct()
              .field("state", Schema.STRING_SCHEMA)
              .field("state.v2", Schema.OPTIONAL_STRING_SCHEMA)
              .build();
      public static final Schema TASK_COUNT_RECORD_V0 = SchemaBuilder.struct()
              .field("task-count", Schema.INT32_SCHEMA)
              .build();
      // The key is logically a byte array, but we can't use the JSON converter to (de-)serialize that without a schema.
      // So instead, we base 64-encode it before serializing and decode it after deserializing.
      public static final Schema SESSION_KEY_V0 = SchemaBuilder.struct()
              .field("key", Schema.STRING_SCHEMA)
              .field("algorithm", Schema.STRING_SCHEMA)
              .field("creation-timestamp", Schema.INT64_SCHEMA)
              .build();
  
      public static final String RESTART_PREFIX = "restart-connector-";
  
      public static String RESTART_KEY(String connectorName) {
          return RESTART_PREFIX + connectorName;
      }
  
      public static final boolean ONLY_FAILED_DEFAULT = false;
      public static final boolean INCLUDE_TASKS_DEFAULT = false;
      public static final String ONLY_FAILED_FIELD_NAME = "only-failed";
      public static final String INCLUDE_TASKS_FIELD_NAME = "include-tasks";
      public static final Schema RESTART_REQUEST_V0 = SchemaBuilder.struct()
              .field(INCLUDE_TASKS_FIELD_NAME, Schema.BOOLEAN_SCHEMA)
              .field(ONLY_FAILED_FIELD_NAME, Schema.BOOLEAN_SCHEMA)
              .build();
  
      public static final String LOGGER_CLUSTER_PREFIX = "logger-cluster-";
      public static String LOGGER_CLUSTER_KEY(String namespace) {
          return LOGGER_CLUSTER_PREFIX + namespace;
      }
      public static final Schema LOGGER_LEVEL_V0 = SchemaBuilder.struct()
              .field("level", Schema.STRING_SCHEMA)
              .build();
  
      // Visible for testing
      static final long READ_WRITE_TOTAL_TIMEOUT_MS = 30000;
  
      private final Object lock;
      private final Converter converter;
      private volatile boolean started;
      // Although updateListener is not final, it's guaranteed to be visible to any thread after its
      // initialization as long as we always read the volatile variable "started" before we access the listener.
      private ConfigBackingStore.UpdateListener updateListener;
  
      private final Map<String, Object> baseProducerProps;
  
      private final String topic;
      // Data is passed to the log already serialized. We use a converter to handle translating to/from generic Connect
      // format to serialized form
      private KafkaBasedLog<String, byte[]> configLog;
      // Connector -> # of tasks
      final Map<String, Integer> connectorTaskCounts = new HashMap<>();
      // Connector and task configs: name or id -> config map
      final Map<String, Map<String, String>> connectorConfigs = new HashMap<>();
      final Map<ConnectorTaskId, Map<String, String>> taskConfigs = new HashMap<>();
      private final Supplier<TopicAdmin> topicAdminSupplier;
      private final String clientId;
  
      // Set of connectors where we saw a task commit with an incomplete set of task config updates, indicating the data
      // is in an inconsistent state and we cannot safely use them until they have been refreshed.
      final Set<String> inconsistent = new HashSet<>();
      // The most recently read offset. This does not take into account deferred task updates/commits, so we may have
      // outstanding data to be applied.
      private volatile long offset;
      // The most recently read session key, to use for validating internal REST requests.
      private volatile SessionKey sessionKey;
  
      // Connector -> Map[ConnectorTaskId -> Configs]
      // visible for testing
      final Map<String, Map<ConnectorTaskId, Map<String, String>>> deferredTaskUpdates = new HashMap<>();
  
      final Map<String, TargetState> connectorTargetStates = new HashMap<>();
  
      final Map<String, Integer> connectorTaskCountRecords = new HashMap<>();
      final Map<String, Integer> connectorTaskConfigGenerations = new HashMap<>();
      final Map<String, AppliedConnectorConfig> appliedConnectorConfigs = new HashMap<>();
      final Set<String> connectorsPendingFencing = new HashSet<>();
  
      private final WorkerConfigTransformer configTransformer;
  
      private final boolean usesFencableWriter;
      private volatile Producer<String, byte[]> fencableProducer;
      private final Map<String, Object> fencableProducerProps;
      private final Time time;
  
      //VisibleForTesting
      void setConfigLog(KafkaBasedLog<String, byte[]> configLog) {
          this.configLog = configLog;
      }
  
      public KafkaConfigBackingStore(Converter converter, DistributedConfig config, WorkerConfigTransformer configTransformer, Supplier<TopicAdmin> adminSupplier, String clientIdBase) {
          this(converter, config, configTransformer, adminSupplier, clientIdBase, Time.SYSTEM);
      }
  
      KafkaConfigBackingStore(Converter converter, DistributedConfig config, WorkerConfigTransformer configTransformer, Supplier<TopicAdmin> adminSupplier, String clientIdBase, Time time) {
          this.lock = new Object();
          this.started = false;
          this.converter = converter;
          this.offset = -1;
          this.topicAdminSupplier = adminSupplier;
          this.clientId = Objects.requireNonNull(clientIdBase) + "configs";
          this.time = time;
  
          this.baseProducerProps = baseProducerProps(config);
          // By default, Connect disables idempotent behavior for all producers, even though idempotence became
          // default for Kafka producers. This is to ensure Connect continues to work with many Kafka broker versions, including older brokers that do not support
          // idempotent producers or require explicit steps to enable them (e.g. adding the IDEMPOTENT_WRITE ACL to brokers older than 2.8).
          // These settings might change when https://cwiki.apache.org/confluence/display/KAFKA/KIP-318%3A+Make+Kafka+Connect+Source+idempotent
          // gets approved and scheduled for release.
          baseProducerProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "false");
  
          this.fencableProducerProps = fencableProducerProps(config);
  
          this.usesFencableWriter = config.transactionalLeaderEnabled();
          this.topic = config.getString(DistributedConfig.CONFIG_TOPIC_CONFIG);
          if (this.topic == null || this.topic.trim().isEmpty())
              throw new ConfigException("Must specify topic for connector configuration.");
  
          configLog = setupAndCreateKafkaBasedLog(this.topic, config);
          this.configTransformer = configTransformer;
      }
  
      @Override
      public void setUpdateListener(ConfigBackingStore.UpdateListener listener) {
          this.updateListener = listener;
      }
  
      @Override
      public void start() {
          log.info("Starting KafkaConfigBackingStore");
          // Before startup, callbacks are *not* invoked. You can grab a snapshot after starting -- just take care that
          // updates can continue to occur in the background
          try {
              configLog.start();
          } catch (UnsupportedVersionException e) {
              throw new ConnectException(
                      "Enabling exactly-once support for source connectors requires a Kafka broker version that allows "
                              + "admin clients to read consumer offsets. Please either disable the worker's exactly-once "
                              + "support for source connectors, or use a newer Kafka broker version.",
                      e
              );
          }
  
          int partitionCount = configLog.partitionCount();
          if (partitionCount > 1) {
              String msg = String.format("Topic '%s' supplied via the '%s' property is required "
                      + "to have a single partition in order to guarantee consistency of "
                      + "connector configurations, but found %d partitions.",
                      topic, DistributedConfig.CONFIG_TOPIC_CONFIG, partitionCount);
              throw new ConfigException(msg);
          }
  
          started = true;
          log.info("Started KafkaConfigBackingStore");
      }
  
      @Override
      public void stop() {
          log.info("Closing KafkaConfigBackingStore");
  
          relinquishWritePrivileges();
          Utils.closeQuietly(configLog::stop, "KafkaBasedLog for config topic");
  
          log.info("Closed KafkaConfigBackingStore");
      }
  
      @Override
      public void claimWritePrivileges() {
          if (usesFencableWriter && fencableProducer == null) {
              try {
                  fencableProducer = createFencableProducer();
                  fencableProducer.initTransactions();
              } catch (Exception e) {
                  relinquishWritePrivileges();
                  throw new ConnectException("Failed to create and initialize fencable producer for config topic", e);
              }
          }
      }
  
      private Map<String, Object> baseProducerProps(WorkerConfig workerConfig) {
          Map<String, Object> producerProps = new HashMap<>(workerConfig.originals());
          String kafkaClusterId = workerConfig.kafkaClusterId();
          producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
          producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
          producerProps.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, Integer.MAX_VALUE);
          ConnectUtils.addMetricsContextProperties(producerProps, workerConfig, kafkaClusterId);
          return producerProps;
      }
  
      // Visible for testing
      Map<String, Object> fencableProducerProps(DistributedConfig workerConfig) {
          Map<String, Object> result = new HashMap<>(baseProducerProps(workerConfig));
  
          result.put(CommonClientConfigs.CLIENT_ID_CONFIG, clientId + "-leader");
          // Always require producer acks to all to ensure durable writes
          result.put(ProducerConfig.ACKS_CONFIG, "all");
          // We can set this to 5 instead of 1 without risking reordering because we are using an idempotent producer
          result.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);
  
          ConnectUtils.ensureProperty(
                  result, ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true",
                  "for the worker's config topic producer when exactly-once source support is enabled or in preparation to be enabled",
                  false
          );
          ConnectUtils.ensureProperty(
                  result, ProducerConfig.TRANSACTIONAL_ID_CONFIG, workerConfig.transactionalProducerId(),
                  "for the worker's config topic producer when exactly-once source support is enabled or in preparation to be enabled",
                  true
          );
  
          return result;
      }
  
      // Visible in order to be mocked during testing
      Producer<String, byte[]> createFencableProducer() {
          return new KafkaProducer<>(fencableProducerProps);
      }
  
      /**
       * Get a snapshot of the current state of the cluster.
       */
      @Override
      public ClusterConfigState snapshot() {
          synchronized (lock) {
              // Only a shallow copy is performed here; in order to avoid accidentally corrupting the worker's view
              // of the config topic, any nested structures should be copied before making modifications
              return new ClusterConfigState(
                      offset,
                      sessionKey,
                      new HashMap<>(connectorTaskCounts),
                      new HashMap<>(connectorConfigs),
                      new HashMap<>(connectorTargetStates),
                      new HashMap<>(taskConfigs),
                      new HashMap<>(connectorTaskCountRecords),
                      new HashMap<>(connectorTaskConfigGenerations),
                      new HashMap<>(appliedConnectorConfigs),
                      new HashSet<>(connectorsPendingFencing),
                      new HashSet<>(inconsistent),
                      configTransformer
              );
          }
      }
  
      @Override
      public boolean contains(String connector) {
          synchronized (lock) {
              return connectorConfigs.containsKey(connector);
          }
      }
  
      /**
       * Write this connector configuration (and optionally a target state) 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 to write data for
       * @param properties the configuration to write
       * @param targetState the desired target state for the connector; may be {@code null} if no target state change is desired. Note that the default
       *                    target state is {@link TargetState#STARTED} if no target state exists previously
       * @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
       */
      @Override
      public void putConnectorConfig(String connector, Map<String, String> properties, TargetState targetState) {
          log.debug("Writing connector configuration for connector '{}'", connector);
          Struct connectConfig = new Struct(CONNECTOR_CONFIGURATION_V0);
          connectConfig.put("properties", properties);
          byte[] serializedConfig = converter.fromConnectData(topic, CONNECTOR_CONFIGURATION_V0, connectConfig);
          try {
              Timer timer = time.timer(READ_WRITE_TOTAL_TIMEOUT_MS);
              List<ProducerKeyValue> keyValues = new ArrayList<>();
              if (targetState != null) {
                  log.debug("Writing target state {} for connector {}", targetState, connector);
                  keyValues.add(new ProducerKeyValue(TARGET_STATE_KEY(connector), serializeTargetState(targetState)));
              }
              keyValues.add(new ProducerKeyValue(CONNECTOR_KEY(connector), serializedConfig));
              sendPrivileged(keyValues, timer);
              configLog.readToEnd().get(timer.remainingMs(), TimeUnit.MILLISECONDS);
          } catch (InterruptedException | ExecutionException | TimeoutException e) {
              log.error("Failed to write connector configuration to Kafka: ", e);
              throw new ConnectException("Error writing connector configuration to Kafka", e);
          }
      }
  
      /**
       * Remove configuration for a given connector. {@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 to remove
       * @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
       */
      @Override
      public void removeConnectorConfig(String connector) {
          log.debug("Removing connector configuration for connector '{}'", connector);
          try {
              Timer timer = time.timer(READ_WRITE_TOTAL_TIMEOUT_MS);
              List<ProducerKeyValue> keyValues = Arrays.asList(
                      new ProducerKeyValue(CONNECTOR_KEY(connector), null),
                      new ProducerKeyValue(TARGET_STATE_KEY(connector), null)
              );
              sendPrivileged(keyValues, timer);
  
              configLog.readToEnd().get(timer.remainingMs(), TimeUnit.MILLISECONDS);
          } catch (InterruptedException | ExecutionException | TimeoutException e) {
              log.error("Failed to remove connector configuration from Kafka: ", e);
              throw new ConnectException("Error removing connector configuration from Kafka", e);
          }
      }
  
      @Override
      public void removeTaskConfigs(String connector) {
          throw new UnsupportedOperationException("Removal of tasks is not currently supported");
      }
  
      /**
       * Write these task configurations and associated commit messages, unless an inconsistency is found that indicates
       * that we would be leaving one of the referenced connectors with an inconsistent state. {@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 the connector to write task configuration
       * @param configs list of task configurations for the connector
       * @throws ConnectException if the task configurations do not resolve inconsistencies found in the existing root
       *                          and task configurations.
       * @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
       */
      @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);
          }
      }
  
      @Override
      public void refresh(long timeout, TimeUnit unit) throws TimeoutException {
          try {
              configLog.readToEnd().get(timeout, unit);
          } catch (InterruptedException | ExecutionException e) {
              throw new ConnectException("Error trying to read to end of config log", e);
          }
      }
  
      /**
       * Write a new {@link TargetState} for the connector. Note that {@link #claimWritePrivileges()} does not need to be
       * invoked before invoking this method.
       * @param connector the name of the connector
       * @param state the desired target state for the connector
       */
      @Override
      public void putTargetState(String connector, TargetState state) {
          log.debug("Writing target state {} for connector {}", state, connector);
          try {
              configLog.sendWithReceipt(TARGET_STATE_KEY(connector), serializeTargetState(state))
                  .get(READ_WRITE_TOTAL_TIMEOUT_MS, TimeUnit.MILLISECONDS);
          } catch (InterruptedException | ExecutionException | TimeoutException e) {
              log.error("Failed to write target state to Kafka", e);
              throw new ConnectException("Error writing target state to Kafka", e);
          }
      }
  
      private byte[] serializeTargetState(TargetState state) {
          Struct connectTargetState = new Struct(TARGET_STATE_V1);
          // Older workers don't support the STOPPED state; fall back on PAUSED
          connectTargetState.put("state", state == STOPPED ? PAUSED.name() : state.name());
          connectTargetState.put("state.v2", state.name());
          return converter.fromConnectData(topic, TARGET_STATE_V1, connectTargetState);
      }
  
      /**
       * 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
       */
      @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 session key 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 sessionKey the session key to distributed
       * @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
       */
      @Override
      public void putSessionKey(SessionKey sessionKey) {
          log.debug("Distributing new session key");
          Struct sessionKeyStruct = new Struct(SESSION_KEY_V0);
          sessionKeyStruct.put("key", Base64.getEncoder().encodeToString(sessionKey.key().getEncoded()));
          sessionKeyStruct.put("algorithm", sessionKey.key().getAlgorithm());
          sessionKeyStruct.put("creation-timestamp", sessionKey.creationTimestamp());
          byte[] serializedSessionKey = converter.fromConnectData(topic, SESSION_KEY_V0, sessionKeyStruct);
          try {
              Timer timer = time.timer(READ_WRITE_TOTAL_TIMEOUT_MS);
              sendPrivileged(SESSION_KEY_KEY, serializedSessionKey, timer);
              configLog.readToEnd().get(timer.remainingMs(), TimeUnit.MILLISECONDS);
          } catch (InterruptedException | ExecutionException | TimeoutException e) {
              log.error("Failed to write session key to Kafka: ", e);
              throw new ConnectException("Error writing session key to Kafka", e);
          }
      }
  
      /**
       * Write a restart request for the connector and optionally its tasks 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 restartRequest the restart request details
       */
      @Override
      public void putRestartRequest(RestartRequest restartRequest) {
          log.debug("Writing {} to Kafka", restartRequest);
          String key = RESTART_KEY(restartRequest.connectorName());
          Struct value = new Struct(RESTART_REQUEST_V0);
          value.put(INCLUDE_TASKS_FIELD_NAME, restartRequest.includeTasks());
          value.put(ONLY_FAILED_FIELD_NAME, restartRequest.onlyFailed());
          byte[] serializedValue = converter.fromConnectData(topic, value.schema(), value);
          try {
              Timer timer = time.timer(READ_WRITE_TOTAL_TIMEOUT_MS);
              sendPrivileged(key, serializedValue, timer);
              configLog.readToEnd().get(timer.remainingMs(), TimeUnit.MILLISECONDS);
          } catch (InterruptedException | ExecutionException | TimeoutException e) {
              log.error("Failed to write {} to Kafka: ", restartRequest, e);
              throw new ConnectException("Error writing " + restartRequest + " to Kafka", e);
          }
      }
  
      @Override
      public void putLoggerLevel(String namespace, String level) {
          log.debug("Writing level {} for logging namespace {} to Kafka", level, namespace);
          Struct value = new Struct(LOGGER_LEVEL_V0);
          value.put("level", level);
          byte[] serializedValue = converter.fromConnectData(topic, value.schema(), value);
          try {
              configLog.sendWithReceipt(LOGGER_CLUSTER_KEY(namespace), serializedValue).get(READ_WRITE_TOTAL_TIMEOUT_MS, TimeUnit.MILLISECONDS);
          } catch (InterruptedException | ExecutionException | TimeoutException e) {
              log.error("Failed to write logger level to Kafka", e);
              throw new ConnectException("Error writing logger level to Kafka", e);
          }
      }
  
      // package private for testing
      KafkaBasedLog<String, byte[]> setupAndCreateKafkaBasedLog(String topic, final WorkerConfig config) {
          String clusterId = config.kafkaClusterId();
          Map<String, Object> originals = config.originals();
  
          Map<String, Object> producerProps = new HashMap<>(baseProducerProps);
          producerProps.put(CommonClientConfigs.CLIENT_ID_CONFIG, clientId);
  
          Map<String, Object> consumerProps = new HashMap<>(originals);
          consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
          consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
          consumerProps.put(CommonClientConfigs.CLIENT_ID_CONFIG, clientId);
          ConnectUtils.addMetricsContextProperties(consumerProps, config, clusterId);
          if (config.exactlyOnceSourceEnabled()) {
              ConnectUtils.ensureProperty(
                      consumerProps, ConsumerConfig.ISOLATION_LEVEL_CONFIG, IsolationLevel.READ_COMMITTED.toString(),
                      "for the worker's config topic consumer when exactly-once source support is enabled",
                      true
              );
          }
  
          Map<String, Object> adminProps = new HashMap<>(originals);
          ConnectUtils.addMetricsContextProperties(adminProps, config, clusterId);
          adminProps.put(CommonClientConfigs.CLIENT_ID_CONFIG, clientId);
  
          Map<String, Object> topicSettings = config instanceof DistributedConfig
                                              ? ((DistributedConfig) config).configStorageTopicSettings()
                                              : Collections.emptyMap();
          NewTopic topicDescription = TopicAdmin.defineTopic(topic)
                  .config(topicSettings) // first so that we override user-supplied settings as needed
                  .compacted()
                  .partitions(1)
                  .replicationFactor(config.getShort(DistributedConfig.CONFIG_STORAGE_REPLICATION_FACTOR_CONFIG))
                  .build();
  
          return createKafkaBasedLog(topic, producerProps, consumerProps, new ConsumeCallback(), topicDescription, topicAdminSupplier, config, time);
      }
  
      /**
       * Send a single record to the config topic synchronously. Note that {@link #claimWritePrivileges()} must be
       * successfully invoked before calling this method if this store is configured to use a fencable writer.
       * @param key the record key
       * @param value the record value
       * @param timer Timer bounding how long this method can block. The timer is updated before the method returns.
       */
      private void sendPrivileged(String key, byte[] value, Timer timer) throws ExecutionException, InterruptedException, TimeoutException {
          sendPrivileged(Collections.singletonList(new ProducerKeyValue(key, value)), timer);
      }
  
      /**
       * Send one or more records to the config topic synchronously. Note that {@link #claimWritePrivileges()} must be
       * successfully invoked before calling this method if this store is configured to use a fencable writer.
       * @param keyValues the list of producer record key/value pairs
       * @param timer Timer bounding how long this method can block. The timer is updated before the method returns.
       */
      private void sendPrivileged(List<ProducerKeyValue> keyValues, Timer timer) throws ExecutionException, InterruptedException, TimeoutException {
          if (!usesFencableWriter) {
              List<Future<RecordMetadata>> producerFutures = new ArrayList<>();
              keyValues.forEach(
                      keyValue -> producerFutures.add(configLog.sendWithReceipt(keyValue.key, keyValue.value))
              );
  
              timer.update();
              for (Future<RecordMetadata> future : producerFutures) {
                  future.get(timer.remainingMs(), TimeUnit.MILLISECONDS);
                  timer.update();
              }
  
              return;
          }
  
          if (fencableProducer == null) {
              throw new IllegalStateException("Cannot produce to config topic without claiming write privileges first");
          }
  
          try {
              fencableProducer.beginTransaction();
              keyValues.forEach(
                      keyValue -> fencableProducer.send(new ProducerRecord<>(topic, keyValue.key, keyValue.value))
              );
              fencableProducer.commitTransaction();
              timer.update();
          } catch (Exception e) {
              log.warn("Failed to perform fencable send to config topic", e);
              relinquishWritePrivileges();
              throw new PrivilegedWriteException("Failed to perform fencable send to config topic", e);
          }
      }
  
      private record ProducerKeyValue(String key, byte[] value) {
      }
  
      private void relinquishWritePrivileges() {
          if (fencableProducer != null) {
              Utils.closeQuietly(() -> fencableProducer.close(Duration.ZERO), "fencable producer for config topic");
              fencableProducer = null;
          }
      }
  
      @Override
      protected String getTopicConfig() {
          return DistributedConfig.CONFIG_TOPIC_CONFIG;
      }
  
      @Override
      protected String getTopicPurpose() {
          return "connector configurations";
      }
  
      private class ConsumeCallback implements Callback<ConsumerRecord<String, byte[]>> {
          @Override
          public void onCompletion(Throwable error, ConsumerRecord<String, byte[]> record) {
              if (error != null) {
                  log.error("Unexpected in consumer callback for KafkaConfigBackingStore: ", error);
                  return;
              }
  
              final SchemaAndValue value;
              try {
                  value = converter.toConnectData(topic, record.value());
              } catch (DataException e) {
                  log.error("Failed to convert config data to Kafka Connect format: ", e);
                  return;
              }
              // Make the recorded offset match the API used for positions in the consumer -- return the offset of the
              // *next record*, not the last one consumed.
              offset = record.offset() + 1;
  
              if (record.key().startsWith(TARGET_STATE_PREFIX)) {
                  String connectorName = record.key().substring(TARGET_STATE_PREFIX.length());
                  processTargetStateRecord(connectorName, value);
              } else if (record.key().startsWith(CONNECTOR_PREFIX)) {
                  String connectorName = record.key().substring(CONNECTOR_PREFIX.length());
                  processConnectorConfigRecord(connectorName, value);
              } else if (record.key().startsWith(TASK_PREFIX)) {
                  ConnectorTaskId taskId = parseTaskId(record.key());
                  if (taskId == null) {
                      log.error("Ignoring task configuration because {} couldn't be parsed as a task config key", record.key());
                      return;
                  }
                  processTaskConfigRecord(taskId, value);
              } else if (record.key().startsWith(COMMIT_TASKS_PREFIX)) {
                  String connectorName = record.key().substring(COMMIT_TASKS_PREFIX.length());
                  processTasksCommitRecord(connectorName, value);
              } else if (record.key().startsWith(RESTART_PREFIX)) {
                  RestartRequest request = recordToRestartRequest(record, value);
                  // Only notify the listener if this backing store is already successfully started (having caught up the first time)
                  if (request != null && started) {
                      updateListener.onRestartRequest(request);
                  }
              } else if (record.key().startsWith(TASK_COUNT_RECORD_PREFIX)) {
                  String connectorName = record.key().substring(TASK_COUNT_RECORD_PREFIX.length());
                  processTaskCountRecord(connectorName, value);
              } else if (record.key().equals(SESSION_KEY_KEY)) {
                  processSessionKeyRecord(value);
              } else if (record.key().startsWith(LOGGER_CLUSTER_PREFIX)) {
                  String loggingNamespace = record.key().substring(LOGGER_CLUSTER_PREFIX.length());
                  processLoggerLevelRecord(loggingNamespace, value);
              } else {
                  log.error("Discarding config update record with invalid key: {}", record.key());
              }
          }
  
      }
  
      private void processTargetStateRecord(String connectorName, SchemaAndValue value) {
          boolean removed = false;
          boolean stateChanged = true;
          synchronized (lock) {
              if (value.value() == null) {
                  // When connector configs are removed, we also write tombstones for the target state.
                  log.debug("Removed target state for connector {} due to null value in topic.", connectorName);
                  connectorTargetStates.remove(connectorName);
                  removed = true;
  
                  // If for some reason we still have configs for the connector, add back the default
                  // STARTED state to ensure each connector always has a valid target state.
                  if (connectorConfigs.containsKey(connectorName))
                      connectorTargetStates.put(connectorName, TargetState.STARTED);
              } else {
                  if (!(value.value() instanceof Map)) {
                      log.error("Ignoring target state for connector '{}' because it is in the wrong format: {}", connectorName, className(value.value()));
                      return;
                  }
                  @SuppressWarnings("unchecked")
                  Map<String, Object> valueMap = (Map<String, Object>) value.value();
                  Object targetState = valueMap.get("state.v2");
                  if (targetState != null && !(targetState instanceof String)) {
                      log.error("Invalid data for target state for connector '{}': 'state.v2' field should be a String but is {}",
                              connectorName, className(targetState));
                      // We don't return here; it's still possible that there's a value we can use in the older state field
                      targetState = null;
                  }
                  if (targetState == null) {
                      // This record may have been written by an older worker; fall back on the older state field
                      targetState = valueMap.get("state");
                      if (!(targetState instanceof String)) {
                          log.error("Invalid data for target state for connector '{}': 'state' field should be a String but is {}",
                                  connectorName, className(targetState));
                          return;
                      }
                  }
  
                  try {
                      TargetState state = TargetState.valueOf((String) targetState);
                      log.debug("Setting target state for connector '{}' to {}", connectorName, targetState);
                      TargetState prevState = connectorTargetStates.put(connectorName, state);
                      stateChanged = !state.equals(prevState);
                  } catch (IllegalArgumentException e) {
                      log.error("Invalid target state for connector '{}': {}", connectorName, targetState);
                      return;
                  }
              }
          }
  
          // Note that we do not notify the update listener if the target state has been removed.
          // Instead we depend on the removal callback of the connector config itself to notify the worker.
          // We also don't notify the update listener if the connector doesn't exist yet - a scenario that can
          // occur if an explicit initial target state is specified in the connector creation request.
          if (started && !removed && stateChanged && connectorConfigs.containsKey(connectorName))
              updateListener.onConnectorTargetStateChange(connectorName);
      }
  
      private void processConnectorConfigRecord(String connectorName, SchemaAndValue value) {
          boolean removed = false;
          synchronized (lock) {
              if (value.value() == null) {
                  // Connector deletion will be written as a null value
                  processConnectorRemoval(connectorName);
                  log.info("Successfully processed removal of connector '{}'", connectorName);
                  removed = true;
              } else {
                  // Connector configs can be applied and callbacks invoked immediately
                  if (!(value.value() instanceof Map)) {
                      log.error("Ignoring configuration for connector '{}' because it is in the wrong format: {}", connectorName, className(value.value()));
                      return;
                  }
                  @SuppressWarnings("unchecked")
                  Object newConnectorConfig = ((Map<String, Object>) value.value()).get("properties");
                  if (!(newConnectorConfig instanceof Map)) {
                      log.error("Invalid data for config for connector '{}': 'properties' field should be a Map but is {}",
                              connectorName, className(newConnectorConfig));
                      return;
                  }
                  log.debug("Updating configuration for connector '{}'", connectorName);
                  @SuppressWarnings("unchecked")
                  Map<String, String> stringsConnectorConfig = (Map<String, String>) newConnectorConfig;
                  connectorConfigs.put(connectorName, stringsConnectorConfig);
  
                  // Set the initial state of the connector to STARTED, which ensures that any connectors
                  // which were created with 0.9 Connect will be initialized in the STARTED state.
                  if (!connectorTargetStates.containsKey(connectorName))
                      connectorTargetStates.put(connectorName, TargetState.STARTED);
              }
          }
          if (started) {
              if (removed)
                  updateListener.onConnectorConfigRemove(connectorName);
              else
                  updateListener.onConnectorConfigUpdate(connectorName);
          }
      }
  
      private void processTaskConfigRecord(ConnectorTaskId taskId, SchemaAndValue value) {
          synchronized (lock) {
              if (value.value() == null) {
                  log.error("Ignoring task configuration for task {} because it is unexpectedly null", taskId);
                  return;
              }
              if (!(value.value() instanceof Map)) {
                  log.error("Ignoring task configuration for task {} because the value is not a Map but is {}", taskId, className(value.value()));
                  return;
              }
  
              @SuppressWarnings("unchecked")
              Object newTaskConfig = ((Map<String, Object>) value.value()).get("properties");
              if (!(newTaskConfig instanceof Map)) {
                  log.error("Invalid data for config of task {} 'properties' field should be a Map but is {}", taskId, className(newTaskConfig));
                  return;
              }
  
              Map<ConnectorTaskId, Map<String, String>> deferred = deferredTaskUpdates.computeIfAbsent(taskId.connector(), k -> new HashMap<>());
              log.debug("Storing new config for task {}; this will wait for a commit message before the new config will take effect.", taskId);
              @SuppressWarnings("unchecked")
              Map<String, String> stringsTaskConfig = (Map<String, String>) newTaskConfig;
              deferred.put(taskId, stringsTaskConfig);
          }
      }
  
      private void processTasksCommitRecord(String connectorName, SchemaAndValue value) {
          List<ConnectorTaskId> updatedTasks = new ArrayList<>();
          synchronized (lock) {
              // Edge case: connector was deleted before these task configs were published,
              // but compaction took place and both the original connector config and the
              // tombstone message for it have been removed from the config topic
              // We should ignore these task configs
              Map<String, String> appliedConnectorConfig = connectorConfigs.get(connectorName);
              if (appliedConnectorConfig == null) {
                  processConnectorRemoval(connectorName);
                  log.debug(
                          "Ignoring task configs for connector {}; it appears that the connector was deleted previously "
                              + "and that log compaction has since removed any trace of its previous configurations "
                              + "from the config topic",
                          connectorName
                  );
                  return;
              }
  
              // Apply any outstanding deferred task updates for the given connector. Note that just because we
              // encounter a commit message does not mean it will result in consistent output. In particular due to
              // compaction, there may be cases where . For example if we have the following sequence of writes:
              //
              // 1. Write connector "foo"'s config
              // 2. Write connector "foo", task 1's config <-- compacted
              // 3. Write connector "foo", task 2's config
              // 4. Write connector "foo" task commit message
              // 5. Write connector "foo", task 1's config
              // 6. Write connector "foo", task 2's config
              // 7. Write connector "foo" task commit message
              //
              // then when a new worker starts up, if message 2 had been compacted, then when message 4 is applied
              // "foo" will not have a complete set of configs. Only when message 7 is applied will the complete
              // configuration be available. Worse, if the leader died while writing messages 5, 6, and 7 such that
              // only 5 was written, then there may be nothing that will finish writing the configs and get the
              // log back into a consistent state.
              //
              // It is expected that the user of this class (i.e., the Herder) will take the necessary action to
              // resolve this (i.e., get the connector to recommit its configuration). This inconsistent state is
              // exposed in the snapshots provided via ClusterConfigState so they are easy to handle.
              if (!(value.value() instanceof Map)) { // Schema-less, so we get maps instead of structs
                  log.error("Ignoring connector tasks configuration commit for connector '{}' because it is in the wrong format: {}", connectorName, className(value.value()));
                  return;
              }
              Map<ConnectorTaskId, Map<String, String>> deferred = deferredTaskUpdates.get(connectorName);
  
              @SuppressWarnings("unchecked")
              int newTaskCount = intValue(((Map<String, Object>) value.value()).get("tasks"));
  
              // Validate the configs we're supposed to update to ensure we're getting a complete configuration
              // update of all tasks that are expected based on the number of tasks in the commit message.
              Set<Integer> taskIdSet = taskIds(connectorName, deferred);
              if (!completeTaskIdSet(taskIdSet, newTaskCount)) {
                  // Given the logic for writing commit messages, we should only hit this condition due to compacted
                  // historical data, in which case we would not have applied any updates yet and there will be no
                  // task config data already committed for the connector, so we shouldn't have to clear any data
                  // out. All we need to do is add the flag marking it inconsistent.
                  log.debug("We have an incomplete set of task configs for connector '{}' probably due to compaction. So we are not doing anything with the new configuration.", connectorName);
                  inconsistent.add(connectorName);
              } else {
                  if (deferred != null) {
                      taskConfigs.putAll(deferred);
                      updatedTasks.addAll(deferred.keySet());
                      connectorTaskConfigGenerations.compute(connectorName, (ignored, generation) -> generation != null ? generation + 1 : 0);
                  }
                  inconsistent.remove(connectorName);
  
                  appliedConnectorConfigs.put(
                          connectorName,
                          new AppliedConnectorConfig(appliedConnectorConfig)
                  );
              }
              // Always clear the deferred entries, even if we didn't apply them. If they represented an inconsistent
              // update, then we need to see a completely fresh set of configs after this commit message, so we don't
              // want any of these outdated configs
              if (deferred != null)
                  deferred.clear();
  
              connectorTaskCounts.put(connectorName, newTaskCount);
          }
  
          // If task configs appear after the latest task count record, the connector needs a new round of zombie fencing
          // before it can start tasks with these configs
          connectorsPendingFencing.add(connectorName);
          if (started)
              updateListener.onTaskConfigUpdate(updatedTasks);
      }
  
      @SuppressWarnings("unchecked")
      RestartRequest recordToRestartRequest(ConsumerRecord<String, byte[]> record, SchemaAndValue value) {
          String connectorName = record.key().substring(RESTART_PREFIX.length());
          if (!(value.value() instanceof Map)) {
              log.error("Ignoring restart request because the value is not a Map but is {}", className(value.value()));
              return null;
          }
  
          Map<String, Object> valueAsMap = (Map<String, Object>) value.value();
  
          Object failed = valueAsMap.get(ONLY_FAILED_FIELD_NAME);
          boolean onlyFailed;
          if (!(failed instanceof Boolean)) {
              log.warn("Invalid data for restart request '{}' field should be a Boolean but is {}, defaulting to {}", ONLY_FAILED_FIELD_NAME, className(failed), ONLY_FAILED_DEFAULT);
              onlyFailed = ONLY_FAILED_DEFAULT;
          } else {
              onlyFailed = (Boolean) failed;
          }
  
          Object withTasks = valueAsMap.get(INCLUDE_TASKS_FIELD_NAME);
          boolean includeTasks;
          if (!(withTasks instanceof Boolean)) {
              log.warn("Invalid data for restart request '{}' field should be a Boolean but is {}, defaulting to {}", INCLUDE_TASKS_FIELD_NAME, className(withTasks), INCLUDE_TASKS_DEFAULT);
              includeTasks = INCLUDE_TASKS_DEFAULT;
          } else {
              includeTasks = (Boolean) withTasks;
          }
          return new RestartRequest(connectorName, onlyFailed, includeTasks);
      }
  
      private void processTaskCountRecord(String connectorName, SchemaAndValue value) {
          if (!(value.value() instanceof Map)) {
              log.error("Ignoring task count record for connector '{}' because it is in the wrong format: {}",  connectorName, className(value.value()));
              return;
          }
          @SuppressWarnings("unchecked")
          int taskCount = intValue(((Map<String, Object>) value.value()).get("task-count"));
  
          log.debug("Setting task count record for connector '{}' to {}", connectorName, taskCount);
          connectorTaskCountRecords.put(connectorName, taskCount);
          // If a task count record appears after the latest task configs, the connector doesn't need a round of zombie
          // fencing before it can start tasks with the latest configs
          connectorsPendingFencing.remove(connectorName);
      }
  
      private void processSessionKeyRecord(SchemaAndValue value) {
          if (value.value() == null) {
              log.error("Ignoring session key because it is unexpectedly null");
              return;
          }
          if (!(value.value() instanceof Map)) {
              log.error("Ignoring session key because the value is not a Map but is {}", className(value.value()));
              return;
          }
  
          @SuppressWarnings("unchecked")
          Map<String, Object> valueAsMap = (Map<String, Object>) value.value();
  
          Object sessionKey = valueAsMap.get("key");
          if (!(sessionKey instanceof String)) {
              log.error("Invalid data for session key 'key' field should be a String but is {}", className(sessionKey));
              return;
          }
          byte[] key = Base64.getDecoder().decode((String) sessionKey);
  
          Object keyAlgorithm = valueAsMap.get("algorithm");
          if (!(keyAlgorithm instanceof String)) {
              log.error("Invalid data for session key 'algorithm' field should be a String but it is {}", className(keyAlgorithm));
              return;
          }
  
          Object creationTimestamp = valueAsMap.get("creation-timestamp");
          if (!(creationTimestamp instanceof Long)) {
              log.error("Invalid data for session key 'creation-timestamp' field should be a long but it is {}", className(creationTimestamp));
              return;
          }
          KafkaConfigBackingStore.this.sessionKey = new SessionKey(
                  new SecretKeySpec(key, (String) keyAlgorithm),
                  (long) creationTimestamp
          );
  
          if (started)
              updateListener.onSessionKeyUpdate(KafkaConfigBackingStore.this.sessionKey);
      }
  
      private void processLoggerLevelRecord(String namespace, SchemaAndValue value) {
          if (value.value() == null) {
              log.error("Ignoring logging level for namespace {} because it is unexpectedly null", namespace);
              return;
          }
          if (!(value.value() instanceof Map)) {
              log.error("Ignoring logging level for namespace {} because the value is not a Map but is {}", namespace, className(value.value()));
              return;
          }
  
          @SuppressWarnings("unchecked")
          Map<String, Object> valueAsMap = (Map<String, Object>) value.value();
  
          Object level = valueAsMap.get("level");
          if (!(level instanceof String)) {
              log.error("Invalid data for logging level key 'level' field with namespace {}; should be a String but it is {}", namespace, className(level));
              return;
          }
  
          if (started) {
              updateListener.onLoggingLevelUpdate(namespace, (String) level);
          } else {
              // TRACE level since there may be many of these records in the config topic
              log.trace(
                      "Ignoring old logging level {} for namespace {} that was written to the config topic before this worker completed startup",
                      level,
                      namespace
              );
          }
      }
  
      private void processConnectorRemoval(String connectorName) {
          connectorConfigs.remove(connectorName);
          connectorTaskCounts.remove(connectorName);
          taskConfigs.keySet().removeIf(taskId -> taskId.connector().equals(connectorName));
          deferredTaskUpdates.remove(connectorName);
          appliedConnectorConfigs.remove(connectorName);
      }
  
      private ConnectorTaskId parseTaskId(String key) {
          String[] parts = key.split("-");
          if (parts.length < 3) return null;
  
          try {
              int taskNum = Integer.parseInt(parts[parts.length - 1]);
              String connectorName = String.join("-", Arrays.copyOfRange(parts, 1, parts.length - 1));
              return new ConnectorTaskId(connectorName, taskNum);
          } catch (NumberFormatException e) {
              return null;
          }
      }
  
      /**
       * Given task configurations, get a set of integer task IDs for the connector.
       */
      private Set<Integer> taskIds(String connector, Map<ConnectorTaskId, Map<String, String>> configs) {
          Set<Integer> tasks = new TreeSet<>();
          if (configs == null) {
              return tasks;
          }
          for (ConnectorTaskId taskId : configs.keySet()) {
              assert taskId.connector().equals(connector);
              tasks.add(taskId.task());
          }
          return tasks;
      }
  
      private boolean completeTaskIdSet(Set<Integer> idSet, int expectedSize) {
          // Note that we do *not* check for the exact set. This is an important implication of compaction. If we start out
          // with 2 tasks, then reduce to 1, we'll end up with log entries like:
          //
          // 1. Connector "foo" config
          // 2. Connector "foo", task 1 config
          // 3. Connector "foo", task 2 config
          // 4. Connector "foo", commit 2 tasks
          // 5. Connector "foo", task 1 config
          // 6. Connector "foo", commit 1 tasks
          //
          // However, due to compaction we could end up with a log that looks like this:
          //
          // 1. Connector "foo" config
          // 3. Connector "foo", task 2 config
          // 5. Connector "foo", task 1 config
          // 6. Connector "foo", commit 1 tasks
          //
          // which isn't incorrect, but would appear in this code to have an extra task configuration. Instead, we just
          // validate that all the configs specified by the commit message are present. This should be fine because the
          // logic for writing configs ensures all the task configs are written (and reads them back) before writing the
          // commit message.
  
          if (idSet.size() < expectedSize)
              return false;
  
          for (int i = 0; i < expectedSize; i++)
              if (!idSet.contains(i))
                  return false;
          return true;
      }
  
      // Convert an integer value extracted from a schemaless struct to an int. This handles potentially different
      // encodings by different Converters.
      private static int intValue(Object value) {
          if (value instanceof Integer)
              return (int) value;
          else if (value instanceof Long)
              return (int) (long) value;
          else
              throw new ConnectException("Expected integer value to be either Integer or Long");
      }
  
  }
  