public boolean verifyTopicCleanupPolicyOnlyCompact(String topic, String workerTopicConfig,
            String topicPurpose) {
        Set<String> cleanupPolicies = topicCleanupPolicy(topic);
        if (cleanupPolicies.isEmpty()) {
            log.info("Unable to use admin client to verify the cleanup policy of '{}' "
                      + "topic is '{}', either because the broker is an older "
                      + "version or because the Kafka principal used for Connect "
                      + "internal topics does not have the required permission to "
                      + "describe topic configurations.", topic, TopicConfig.CLEANUP_POLICY_COMPACT);
            return false;
        }
        Set<String> expectedPolicies = Collections.singleton(TopicConfig.CLEANUP_POLICY_COMPACT);
        if (!cleanupPolicies.equals(expectedPolicies)) {
            String expectedPolicyStr = String.join(",", expectedPolicies);
            String cleanupPolicyStr = String.join(",", cleanupPolicies);
            String msg = String.format("Topic '%s' supplied via the '%s' property is required "
                    + "to have '%s=%s' to guarantee consistency and durability of "
                    + "%s, but found the topic currently has '%s=%s'. Continuing would likely "
                    + "result in eventually losing %s and problems restarting this Connect "
                    + "cluster in the future. Change the '%s' property in the "
                    + "Connect worker configurations to use a topic with '%s=%s'.",
                    topic, workerTopicConfig, TopicConfig.CLEANUP_POLICY_CONFIG, expectedPolicyStr,
                    topicPurpose, TopicConfig.CLEANUP_POLICY_CONFIG, cleanupPolicyStr, topicPurpose,
                    workerTopicConfig, TopicConfig.CLEANUP_POLICY_CONFIG, expectedPolicyStr);
            throw new ConfigException(msg);
        }
        return true;
    }
Verify the named topic uses only compaction for the cleanup policy.

@param topic             the name of the topic
@param workerTopicConfig the name of the worker configuration that specifies the topic name
@return true if the admin client could be used to verify the topic setting, or false if
        the verification could not be performed, likely because the admin client principal
        did not have the required permissions or because the broker was older than 0.11.0.0
@throws ConfigException if the actual topic setting did not match the required setting
connect/runtime/src/main/java/org/apache/kafka/connect/util/TopicAdmin.java