@Test
    public void testCompactedDeletedOlderConnectorConfig() throws Exception {
        brokerProps.put("log.cleaner.backoff.ms", "100");
        brokerProps.put("log.cleaner.delete.retention.ms", "1");
        brokerProps.put("log.cleaner.max.compaction.lag.ms", "1");
        brokerProps.put("log.cleaner.min.cleanable.ratio", "0");
        brokerProps.put("log.cleaner.min.compaction.lag.ms", "1");
        brokerProps.put("log.cleaner.threads", "1");

        final String configTopic = "kafka-16838-configs";
        final int offsetCommitIntervalMs = 100;
        workerProps.put(CONFIG_TOPIC_CONFIG, configTopic);
        workerProps.put(CONFIG_STORAGE_PREFIX + SEGMENT_MS_CONFIG, "100");
        workerProps.put(CONFIG_STORAGE_PREFIX + DELETE_RETENTION_MS_CONFIG, "1");
        workerProps.put(OFFSET_COMMIT_INTERVAL_MS_CONFIG, Integer.toString(offsetCommitIntervalMs));

        final int numWorkers = 1;
        connect = connectBuilder
                .numWorkers(numWorkers)
                .build();
        // start the clusters
        connect.start();

        final String connectorTopic = "connector-topic";
        connect.kafka().createTopic(connectorTopic, 1);

        ConnectorHandle connectorHandle = RuntimeHandles.get().connectorHandle(CONNECTOR_NAME);
        connectorHandle.expectedCommits(NUM_TASKS * 2);

        Map<String, String> connectorConfig = defaultSourceConnectorProps(connectorTopic);
        connect.configureConnector(CONNECTOR_NAME, connectorConfig);
        connect.assertions().assertConnectorAndExactlyNumTasksAreRunning(
                CONNECTOR_NAME,
                NUM_TASKS,
                "Connector or its tasks did not start in time"
        );
        connectorHandle.awaitCommits(RECORD_TRANSFER_TIMEOUT_MS);

        connect.deleteConnector(CONNECTOR_NAME);

        // Roll the entire cluster
        connect.healthyWorkers().forEach(connect::removeWorker);

        // Miserable hack: produce directly to the config topic and then wait a little bit
        // in order to trigger segment rollover and allow compaction to take place
        connect.kafka().produce(configTopic, "garbage-key-1", null);
        Thread.sleep(1_000);
        connect.kafka().produce(configTopic, "garbage-key-2", null);
        Thread.sleep(1_000);

        for (int i = 0; i < numWorkers; i++)
            connect.addWorker();

        connect.assertions().assertAtLeastNumWorkersAreUp(
                numWorkers,
                "Workers did not start in time after cluster was rolled."
        );

        final TopicPartition connectorTopicPartition = new TopicPartition(connectorTopic, 0);
        final long initialEndOffset = connect.kafka().endOffset(connectorTopicPartition);
        assertTrue(
                initialEndOffset > 0,
                "Source connector should have published at least one record to Kafka"
        );

        connectorHandle.expectedCommits(NUM_TASKS * 2);

        // Re-create the connector with a different config (targets a different topic)
        final String otherConnectorTopic = "other-topic";
        connect.kafka().createTopic(otherConnectorTopic, 1);
        connectorConfig.put(TOPIC_CONFIG, otherConnectorTopic);
        connect.configureConnector(CONNECTOR_NAME, connectorConfig);
        connect.assertions().assertConnectorAndExactlyNumTasksAreRunning(
                CONNECTOR_NAME,
                NUM_TASKS,
                "Connector or its tasks did not start in time"
        );
        connectorHandle.awaitCommits(RECORD_TRANSFER_TIMEOUT_MS);

        // See if any new records got written to the old topic
        final long nextEndOffset = connect.kafka().endOffset(connectorTopicPartition);
        assertEquals(
                initialEndOffset,
                nextEndOffset,
                "No new records should have been written to the older topic"
        );
    }
Task configs are not removed from the config topic after a connector is deleted.
When topic compaction takes place, this can cause the tombstone message for the
connector config to be deleted, leaving the task configs in the config topic with no
explicit record of the connector's deletion.
<p>
This test guarantees that those older task configs are never used, even when the
connector is recreated later.
connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectWorkerIntegrationTest.java