@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);
    }
@Test
    public void misconfigurationOfStatusBackingStore() {
        when(workerConfig.getString(DistributedConfig.STATUS_STORAGE_TOPIC_CONFIG)).thenReturn(null);
        when(workerConfig.getString(DistributedConfig.STATUS_STORAGE_TOPIC_CONFIG)).thenReturn("   ");

        Exception e = assertThrows(ConfigException.class, () -> store.configure(workerConfig));
        assertEquals("Must specify topic for connector status.", e.getMessage());
        e = assertThrows(ConfigException.class, () -> store.configure(workerConfig));
        assertEquals("Must specify topic for connector status.", e.getMessage());
    }