private void readTopicStatus(String key, byte[] value) {
        int delimiterPos = key.indexOf(':');
        int beginPos = TOPIC_STATUS_PREFIX.length();
        if (beginPos > delimiterPos) {
            log.warn("Discarding record with invalid topic status key {}", key);
            return;
        }

        String topic = key.substring(beginPos, delimiterPos);
        if (topic.isEmpty()) {
            log.warn("Discarding record with invalid topic status key containing empty topic {}", key);
            return;
        }

        beginPos = delimiterPos + TOPIC_STATUS_SEPARATOR.length();
        int endPos = key.length();
        if (beginPos > endPos) {
            log.warn("Discarding record with invalid topic status key {}", key);
            return;
        }

        String connector = key.substring(beginPos);
        if (connector.isEmpty()) {
            log.warn("Discarding record with invalid topic status key containing empty connector {}", key);
            return;
        }

        if (value == null) {
            log.trace("Removing status for topic {} and connector {}", topic, connector);
            removeTopic(topic, connector);
            return;
        }

        TopicStatus status = parseTopicStatus(value);
        if (status == null) {
            log.warn("Failed to parse topic status with key {}", key);
            return;
        }

        log.trace("Received topic status update {}", status);
        topics.computeIfAbsent(connector, k -> new ConcurrentHashMap<>())
                .put(topic, status);
    }
StatusBackingStore implementation which uses a compacted topic for storage
of connector and task status information. When a state change is observed,
the new state is written to the compacted topic. The new state will not be
visible until it has been read back from the topic.
<p>
In spite of their names, the {@link #putSafe} methods cannot guarantee the safety
of the write (since Kafka itself cannot provide such guarantees currently),
but it can avoid specific unsafe conditions. In particular, {@link #putSafe}
allows writes in the following conditions:
<ol>
  <li>It is (probably) safe to overwrite the state if there is no previous
  value.
  <li>It is (probably) safe to overwrite the state if the previous value was
  set by a worker with the same workerId.
  <li>It is (probably) safe to overwrite the previous state if the current
  generation is higher than the previous .
</ol>
Basically all these conditions do is reduce the window for conflicts. They
obviously cannot take into account in-flight requests.
connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaStatusBackingStore.java