protected TopicStatus parseTopicStatus(byte[] data) {
        try {
            SchemaAndValue schemaAndValue = converter.toConnectData(statusTopic, data);
            if (!(schemaAndValue.value() instanceof Map)) {
                log.error("Invalid topic status value {}", schemaAndValue.value());
                return null;
            }
            @SuppressWarnings("unchecked")
            Object innerValue = ((Map<String, Object>) schemaAndValue.value()).get(TOPIC_STATE_KEY);
            if (!(innerValue instanceof Map)) {
                log.error("Invalid topic status value {} for field {}", innerValue, TOPIC_STATE_KEY);
                return null;
            }
            @SuppressWarnings("unchecked")
            Map<String, Object> topicStatusMetadata = (Map<String, Object>) innerValue;
            return new TopicStatus((String) topicStatusMetadata.get(TOPIC_NAME_KEY),
                    (String) topicStatusMetadata.get(TOPIC_CONNECTOR_KEY),
                    ((Long) topicStatusMetadata.get(TOPIC_TASK_KEY)).intValue(),
                    (long) topicStatusMetadata.get(TOPIC_DISCOVER_TIMESTAMP_KEY));
        } catch (Exception e) {
            log.error("Failed to deserialize topic status", e);
            return null;
        }
    }
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