private void readTaskStatus(String key, byte[] value) {
        ConnectorTaskId id = parseConnectorTaskId(key);
        if (id == null) {
            log.warn("Discarding record with invalid task status key {}", key);
            return;
        }

        if (value == null) {
            log.trace("Removing task status for {}", id);
            remove(id);
            return;
        }

        TaskStatus status = parseTaskStatus(id, value);
        if (status == null) {
            log.warn("Failed to parse task status with key {}", key);
            return;
        }

        synchronized (this) {
            log.trace("Received task {} status update {}", id, status);
            CacheEntry<TaskStatus> entry = getOrAdd(id);

            // During frequent rebalances, there could be a race condition because of which
            // an UNASSIGNED state of a prior or same generation can be sent by a worker despite a
            // RUNNING status by another worker because the first worker
            // couldn't read the latest RUNNING status. This can lead to an inaccurate status
            // representation even though the task might be actually running.
            // Note that this could also mean that when a generation reset happens, and an
            // UNASSIGNED status is sent, then it would be ignored if the current status is RUNNING
            // at a higher generation. But since it will be followed by a RUNNING or a different
            // status message(at the lower generation) soon after, the misrepresentation of the UNASSIGNED
            // status would be short-lived in most cases.
            if (status.state() == TaskStatus.State.UNASSIGNED
                    && entry.get() != null
                    && entry.get().state() == TaskStatus.State.RUNNING
                    && entry.get().generation() >= status.generation()) {
                log.trace("Ignoring stale status {} in favour of more upto date status {}", status, entry.get());
                return;
            }

            entry.put(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