/*
 * 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.ProducerConfig;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.config.ConfigException;
import org.apache.kafka.common.errors.RetriableException;
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.ThreadUtils;
import org.apache.kafka.common.utils.Time;
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.runtime.AbstractStatus;
import org.apache.kafka.connect.runtime.ConnectorStatus;
import org.apache.kafka.connect.runtime.TaskStatus;
import org.apache.kafka.connect.runtime.TopicStatus;
import org.apache.kafka.connect.runtime.WorkerConfig;
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.Table;
import org.apache.kafka.connect.util.TopicAdmin;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
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.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;

/**
 * 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.
 *
 */
public class KafkaStatusBackingStore extends KafkaTopicBasedBackingStore implements StatusBackingStore {
    private static final Logger log = LoggerFactory.getLogger(KafkaStatusBackingStore.class);

    public static final String TASK_STATUS_PREFIX = "status-task-";
    public static final String CONNECTOR_STATUS_PREFIX = "status-connector-";
    public static final String TOPIC_STATUS_PREFIX = "status-topic-";
    public static final String TOPIC_STATUS_SEPARATOR = ":connector-";

    public static final String STATE_KEY_NAME = "state";
    public static final String TRACE_KEY_NAME = "trace";
    public static final String WORKER_ID_KEY_NAME = "worker_id";
    public static final String GENERATION_KEY_NAME = "generation";

    public static final String TOPIC_STATE_KEY = "topic";
    public static final String TOPIC_NAME_KEY = "name";
    public static final String TOPIC_CONNECTOR_KEY = "connector";
    public static final String TOPIC_TASK_KEY = "task";
    public static final String TOPIC_DISCOVER_TIMESTAMP_KEY = "discoverTimestamp";

    private static final Schema STATUS_SCHEMA_V0 = SchemaBuilder.struct()
            .field(STATE_KEY_NAME, Schema.STRING_SCHEMA)
            .field(TRACE_KEY_NAME, SchemaBuilder.string().optional().build())
            .field(WORKER_ID_KEY_NAME, Schema.STRING_SCHEMA)
            .field(GENERATION_KEY_NAME, Schema.INT32_SCHEMA)
            .build();

    private static final Schema TOPIC_STATUS_VALUE_SCHEMA_V0 = SchemaBuilder.struct()
            .field(TOPIC_NAME_KEY, Schema.STRING_SCHEMA)
            .field(TOPIC_CONNECTOR_KEY, Schema.STRING_SCHEMA)
            .field(TOPIC_TASK_KEY, Schema.INT32_SCHEMA)
            .field(TOPIC_DISCOVER_TIMESTAMP_KEY, Schema.INT64_SCHEMA)
            .build();

    private static final Schema TOPIC_STATUS_SCHEMA_V0 = SchemaBuilder.map(
            Schema.STRING_SCHEMA,
            TOPIC_STATUS_VALUE_SCHEMA_V0
    ).build();

    private final Time time;
    private final Converter converter;
    //visible for testing
    protected final Table<String, Integer, CacheEntry<TaskStatus>> tasks;
    protected final Map<String, CacheEntry<ConnectorStatus>> connectors;
    protected final ConcurrentMap<String, ConcurrentMap<String, TopicStatus>> topics;
    private final Supplier<TopicAdmin> topicAdminSupplier;
    private final String clientId;

    private String statusTopic;
    private KafkaBasedLog<String, byte[]> kafkaLog;
    private int generation;
    private ExecutorService sendRetryExecutor;

    public KafkaStatusBackingStore(Time time, Converter converter, Supplier<TopicAdmin> topicAdminSupplier, String clientIdBase) {
        this.time = time;
        this.converter = converter;
        this.tasks = new Table<>();
        this.connectors = new HashMap<>();
        this.topics = new ConcurrentHashMap<>();
        this.topicAdminSupplier = topicAdminSupplier;
        this.clientId = Objects.requireNonNull(clientIdBase) + "statuses";
    }

    // visible for testing
    KafkaStatusBackingStore(Time time, Converter converter, String statusTopic, Supplier<TopicAdmin> topicAdminSupplier,
        KafkaBasedLog<String, byte[]> kafkaLog) {
        this(time, converter, null, "connect-distributed-");
        this.kafkaLog = kafkaLog;
        this.statusTopic = statusTopic;
        sendRetryExecutor = Executors.newSingleThreadExecutor(
                ThreadUtils.createThreadFactory("status-store-retry-" + statusTopic, true));
    }

    @Override
    public void configure(final WorkerConfig config) {
        this.statusTopic = config.getString(DistributedConfig.STATUS_STORAGE_TOPIC_CONFIG);
        if (this.statusTopic == null || this.statusTopic.trim().isEmpty())
            throw new ConfigException("Must specify topic for connector status.");

        sendRetryExecutor = Executors.newSingleThreadExecutor(
                ThreadUtils.createThreadFactory("status-store-retry-" + statusTopic, true));

        String clusterId = config.kafkaClusterId();
        Map<String, Object> originals = config.originals();
        Map<String, Object> producerProps = new HashMap<>(originals);
        producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
        producerProps.put(ProducerConfig.RETRIES_CONFIG, 0); // we handle retries in this class
        // 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.
        producerProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "false"); // disable idempotence since retries is force to 0
        producerProps.put(CommonClientConfigs.CLIENT_ID_CONFIG, clientId);
        ConnectUtils.addMetricsContextProperties(producerProps, config, clusterId);

        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);

        Map<String, Object> adminProps = new HashMap<>(originals);
        adminProps.put(CommonClientConfigs.CLIENT_ID_CONFIG, clientId);
        ConnectUtils.addMetricsContextProperties(adminProps, config, clusterId);

        Map<String, Object> topicSettings = config instanceof DistributedConfig
                                            ? ((DistributedConfig) config).statusStorageTopicSettings()
                                            : Collections.emptyMap();
        NewTopic topicDescription = TopicAdmin.defineTopic(statusTopic)
                .config(topicSettings) // first so that we override user-supplied settings as needed
                .compacted()
                .partitions(config.getInt(DistributedConfig.STATUS_STORAGE_PARTITIONS_CONFIG))
                .replicationFactor(config.getShort(DistributedConfig.STATUS_STORAGE_REPLICATION_FACTOR_CONFIG))
                .build();

        Callback<ConsumerRecord<String, byte[]>> readCallback = (error, record) -> read(record);
        this.kafkaLog = createKafkaBasedLog(statusTopic, producerProps, consumerProps, readCallback, topicDescription, topicAdminSupplier, config, time);
    }

    @Override
    public void start() {
        kafkaLog.start();

        // read to the end on startup to ensure that api requests see the most recent states
        kafkaLog.readToEnd();
    }

    @Override
    public void stop() {
        try {
            kafkaLog.stop();
        } finally {
            ThreadUtils.shutdownExecutorServiceQuietly(sendRetryExecutor, 10, TimeUnit.SECONDS);
        }
    }

    @Override
    public void put(final ConnectorStatus status) {
        sendConnectorStatus(status, false);
    }

    @Override
    public void putSafe(final ConnectorStatus status) {
        sendConnectorStatus(status, true);
    }

    @Override
    public void put(final TaskStatus status) {
        sendTaskStatus(status, false);
    }

    @Override
    public void putSafe(final TaskStatus status) {
        sendTaskStatus(status, true);
    }

    @Override
    public void put(final TopicStatus status) {
        sendTopicStatus(status.connector(), status.topic(), status);
    }

    @Override
    public void flush() {
        kafkaLog.flush();
    }

    private void sendConnectorStatus(final ConnectorStatus status, boolean safeWrite) {
        String connector = status.id();
        CacheEntry<ConnectorStatus> entry = getOrAdd(connector);
        String key = CONNECTOR_STATUS_PREFIX + connector;
        send(key, status, entry, safeWrite);
    }

    private void sendTaskStatus(final TaskStatus status, boolean safeWrite) {
        ConnectorTaskId taskId = status.id();
        CacheEntry<TaskStatus> entry = getOrAdd(taskId);
        String key = TASK_STATUS_PREFIX + taskId.connector() + "-" + taskId.task();
        send(key, status, entry, safeWrite);
    }

    private void sendTopicStatus(final String connector, final String topic, final TopicStatus status) {
        String key = TOPIC_STATUS_PREFIX + topic + TOPIC_STATUS_SEPARATOR + connector;

        final byte[] value = serializeTopicStatus(status);

        kafkaLog.send(key, value, new org.apache.kafka.clients.producer.Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                if (exception == null) return;
                // TODO: retry more gracefully and not forever
                if (exception instanceof RetriableException) {
                    sendRetryExecutor.submit(() -> kafkaLog.send(key, value, this));
                } else {
                    log.error("Failed to write status update", exception);
                }
            }
        });
    }

    private <V extends AbstractStatus<?>> void send(final String key,
                                                 final V status,
                                                 final CacheEntry<V> entry,
                                                 final boolean safeWrite) {
        final int sequence;
        synchronized (this) {
            this.generation = status.generation();
            if (safeWrite && !entry.canWriteSafely(status))
                return;
            sequence = entry.increment();
        }

        final byte[] value = status.state() == ConnectorStatus.State.DESTROYED ? null : serialize(status);

        kafkaLog.send(key, value, new org.apache.kafka.clients.producer.Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                if (exception == null) return;
                if (exception instanceof RetriableException) {
                    synchronized (KafkaStatusBackingStore.this) {
                        if (entry.isDeleted()
                            || status.generation() != generation
                            || (safeWrite && !entry.canWriteSafely(status, sequence)))
                            return;
                    }

                    sendRetryExecutor.submit(() -> kafkaLog.send(key, value, this));
                } else {
                    log.error("Failed to write status update", exception);
                }
            }
        });
    }

    private synchronized CacheEntry<ConnectorStatus> getOrAdd(String connector) {
        CacheEntry<ConnectorStatus> entry = connectors.get(connector);
        if (entry == null) {
            entry = new CacheEntry<>();
            connectors.put(connector, entry);
        }
        return entry;
    }

    private synchronized void remove(String connector) {
        CacheEntry<ConnectorStatus> removed = connectors.remove(connector);
        if (removed != null)
            removed.delete();

        Map<Integer, CacheEntry<TaskStatus>> tasks = this.tasks.remove(connector);
        if (tasks != null) {
            for (CacheEntry<TaskStatus> taskEntry : tasks.values())
                taskEntry.delete();
        }
    }

    private synchronized CacheEntry<TaskStatus> getOrAdd(ConnectorTaskId task) {
        CacheEntry<TaskStatus> entry = tasks.get(task.connector(), task.task());
        if (entry == null) {
            entry = new CacheEntry<>();
            tasks.put(task.connector(), task.task(), entry);
        }
        return entry;
    }

    private synchronized void remove(ConnectorTaskId id) {
        CacheEntry<TaskStatus> removed = tasks.remove(id.connector(), id.task());
        if (removed != null)
            removed.delete();
    }

    private void removeTopic(String topic, String connector) {
        ConcurrentMap<String, TopicStatus> activeTopics = topics.get(connector);
        if (activeTopics == null) {
            return;
        }
        activeTopics.remove(topic);
    }

    @Override
    public synchronized TaskStatus get(ConnectorTaskId id) {
        CacheEntry<TaskStatus> entry = tasks.get(id.connector(), id.task());
        return entry == null ? null : entry.get();
    }

    @Override
    public synchronized ConnectorStatus get(String connector) {
        CacheEntry<ConnectorStatus> entry = connectors.get(connector);
        return entry == null ? null : entry.get();
    }

    @Override
    public synchronized Collection<TaskStatus> getAll(String connector) {
        List<TaskStatus> res = new ArrayList<>();
        for (CacheEntry<TaskStatus> statusEntry : tasks.row(connector).values()) {
            TaskStatus status = statusEntry.get();
            if (status != null)
                res.add(status);
        }
        return res;
    }

    @Override
    public TopicStatus getTopic(String connector, String topic) {
        ConcurrentMap<String, TopicStatus> activeTopics = topics.get(Objects.requireNonNull(connector));
        return activeTopics != null ? activeTopics.get(Objects.requireNonNull(topic)) : null;
    }

    @Override
    public Collection<TopicStatus> getAllTopics(String connector) {
        ConcurrentMap<String, TopicStatus> activeTopics = topics.get(Objects.requireNonNull(connector));
        return activeTopics != null
               ? Collections.unmodifiableCollection(Objects.requireNonNull(activeTopics.values()))
               : Collections.emptySet();
    }

    @Override
    public void deleteTopic(String connector, String topic) {
        sendTopicStatus(Objects.requireNonNull(connector), Objects.requireNonNull(topic), null);
    }

    @Override
    public synchronized Set<String> connectors() {
        return new HashSet<>(connectors.keySet());
    }

    private ConnectorStatus parseConnectorStatus(String connector, byte[] data) {
        try {
            SchemaAndValue schemaAndValue = converter.toConnectData(statusTopic, data);
            if (!(schemaAndValue.value() instanceof Map)) {
                log.error("Invalid connector status type {}", schemaAndValue.value().getClass());
                return null;
            }

            @SuppressWarnings("unchecked")
            Map<String, Object> statusMap = (Map<String, Object>) schemaAndValue.value();
            TaskStatus.State state = TaskStatus.State.valueOf((String) statusMap.get(STATE_KEY_NAME));
            String trace = (String) statusMap.get(TRACE_KEY_NAME);
            String workerUrl = (String) statusMap.get(WORKER_ID_KEY_NAME);
            int generation = ((Long) statusMap.get(GENERATION_KEY_NAME)).intValue();
            return new ConnectorStatus(connector, state, trace, workerUrl, generation);
        } catch (Exception e) {
            log.error("Failed to deserialize connector status", e);
            return null;
        }
    }

    private TaskStatus parseTaskStatus(ConnectorTaskId taskId, byte[] data) {
        try {
            SchemaAndValue schemaAndValue = converter.toConnectData(statusTopic, data);
            if (!(schemaAndValue.value() instanceof Map)) {
                log.error("Invalid task status type {}", schemaAndValue.value().getClass());
                return null;
            }
            @SuppressWarnings("unchecked")
            Map<String, Object> statusMap = (Map<String, Object>) schemaAndValue.value();
            TaskStatus.State state = TaskStatus.State.valueOf((String) statusMap.get(STATE_KEY_NAME));
            String trace = (String) statusMap.get(TRACE_KEY_NAME);
            String workerUrl = (String) statusMap.get(WORKER_ID_KEY_NAME);
            int generation = ((Long) statusMap.get(GENERATION_KEY_NAME)).intValue();
            return new TaskStatus(taskId, state, workerUrl, generation, trace);
        } catch (Exception e) {
            log.error("Failed to deserialize task status", e);
            return null;
        }
    }

    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;
        }
    }

    private byte[] serialize(AbstractStatus<?> status) {
        Struct struct = new Struct(STATUS_SCHEMA_V0);
        struct.put(STATE_KEY_NAME, status.state().name());
        if (status.trace() != null)
            struct.put(TRACE_KEY_NAME, status.trace());
        struct.put(WORKER_ID_KEY_NAME, status.workerId());
        struct.put(GENERATION_KEY_NAME, status.generation());
        return converter.fromConnectData(statusTopic, STATUS_SCHEMA_V0, struct);
    }

    //visible for testing
    protected byte[] serializeTopicStatus(TopicStatus status) {
        if (status == null) {
            // This should send a tombstone record that will represent delete
            return null;
        }
        Struct struct = new Struct(TOPIC_STATUS_VALUE_SCHEMA_V0);
        struct.put(TOPIC_NAME_KEY, status.topic());
        struct.put(TOPIC_CONNECTOR_KEY, status.connector());
        struct.put(TOPIC_TASK_KEY, status.task());
        struct.put(TOPIC_DISCOVER_TIMESTAMP_KEY, status.discoverTimestamp());
        return converter.fromConnectData(
                statusTopic,
                TOPIC_STATUS_SCHEMA_V0,
                Collections.singletonMap(TOPIC_STATE_KEY, struct));
    }

    private String parseConnectorStatusKey(String key) {
        return key.substring(CONNECTOR_STATUS_PREFIX.length());
    }

    private ConnectorTaskId parseConnectorTaskId(String key) {
        String[] parts = key.split("-");
        if (parts.length < 4) return null;

        try {
            int taskNum = Integer.parseInt(parts[parts.length - 1]);
            String connectorName = String.join("-", Arrays.copyOfRange(parts, 2, parts.length - 1));
            return new ConnectorTaskId(connectorName, taskNum);
        } catch (NumberFormatException e) {
            log.warn("Invalid task status key {}", key);
            return null;
        }
    }

    private void readConnectorStatus(String key, byte[] value) {
        String connector = parseConnectorStatusKey(key);
        if (connector.isEmpty()) {
            log.warn("Discarding record with invalid connector status key {}", key);
            return;
        }

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

        ConnectorStatus status = parseConnectorStatus(connector, value);
        if (status == null)
            return;

        synchronized (this) {
            log.trace("Received connector {} status update {}", connector, status);
            CacheEntry<ConnectorStatus> entry = getOrAdd(connector);
            entry.put(status);
        }
    }

    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);
        }
    }

    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);
    }

    // visible for testing
    void read(ConsumerRecord<String, byte[]> record) {
        String key = record.key();
        if (key.startsWith(CONNECTOR_STATUS_PREFIX)) {
            readConnectorStatus(key, record.value());
        } else if (key.startsWith(TASK_STATUS_PREFIX)) {
            readTaskStatus(key, record.value());
        } else if (key.startsWith(TOPIC_STATUS_PREFIX)) {
            readTopicStatus(key, record.value());
        } else {
            log.warn("Discarding record with invalid key {}", key);
        }
    }

    @Override
    protected String getTopicConfig() {
        return DistributedConfig.STATUS_STORAGE_TOPIC_CONFIG;
    }

    @Override
    protected String getTopicPurpose() {
        return "connector and task statuses";
    }

    private static class CacheEntry<T extends AbstractStatus<?>> {
        private T value = null;
        private int sequence = 0;
        private boolean deleted = false;

        public int increment() {
            return ++sequence;
        }

        public void put(T value) {
            this.value = value;
        }

        public T get() {
            return value;
        }

        public void delete() {
            this.deleted = true;
        }

        public boolean isDeleted() {
            return deleted;
        }

        public boolean canWriteSafely(T status) {
            return value == null
                    || value.workerId().equals(status.workerId())
                    || value.generation() <= status.generation();
        }

        public boolean canWriteSafely(T status, int sequence) {
            return canWriteSafely(status) && this.sequence == sequence;
        }

    }

}
