public void assertNoTopicStatusInStatusTopic() {
        String statusTopic = workerProps.get(DistributedConfig.STATUS_STORAGE_TOPIC_CONFIG);
        Consumer<byte[], byte[]> verifiableConsumer = connect.kafka().createConsumer(
                Collections.singletonMap("group.id", "verifiable-consumer-group-0"));

        List<PartitionInfo> partitionInfos = verifiableConsumer.partitionsFor(statusTopic);
        if (partitionInfos.isEmpty()) {
            throw new AssertionError("Unable to retrieve partitions info for status topic");
        }
        List<TopicPartition> partitions = partitionInfos.stream()
                .map(info -> new TopicPartition(info.topic(), info.partition()))
                .collect(Collectors.toList());
        verifiableConsumer.assign(partitions);

        // Based on the implementation of {@link org.apache.kafka.connect.util.KafkaBasedLog#readToLogEnd}
        Set<TopicPartition> assignment = verifiableConsumer.assignment();
        verifiableConsumer.seekToBeginning(assignment);
        Map<TopicPartition, Long> endOffsets = verifiableConsumer.endOffsets(assignment);
        while (!endOffsets.isEmpty()) {
            Iterator<Map.Entry<TopicPartition, Long>> it = endOffsets.entrySet().iterator();
            while (it.hasNext()) {
                Map.Entry<TopicPartition, Long> entry = it.next();
                if (verifiableConsumer.position(entry.getKey()) >= entry.getValue())
                    it.remove();
                else {
                    try {
                        StreamSupport.stream(verifiableConsumer.poll(Duration.ofMillis(Integer.MAX_VALUE)).spliterator(), false)
                                .map(ConsumerRecord::key)
                                .filter(Objects::nonNull)
                                .filter(key -> new String(key, StandardCharsets.UTF_8).startsWith(KafkaStatusBackingStore.TOPIC_STATUS_PREFIX))
                                .findFirst()
                                .ifPresent(key -> {
                                    throw new AssertionError("Found unexpected key: " + new String(key, StandardCharsets.UTF_8) + " in status topic");
                                });
                    } catch (KafkaException e) {
                        throw new AssertionError("Error while reading to the end of status topic", e);
                    }
                    break;
                }
            }
        }
        verifiableConsumer.close();
    }
Integration test for the endpoints that offer topic tracking of a connector's active
topics.
connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectorTopicsIntegrationTest.java