public Set<String> topicCleanupPolicy(String topic) {
        Config topicConfig = describeTopicConfig(topic);
        if (topicConfig == null) {
            // The topic must not exist
            log.debug("Unable to find topic '{}' when getting cleanup policy", topic);
            return Collections.emptySet();
        }
        ConfigEntry entry = topicConfig.get(CLEANUP_POLICY_CONFIG);
        if (entry != null && entry.value() != null) {
            String policyStr = entry.value();
            log.debug("Found cleanup.policy={} for topic '{}'", policyStr, topic);
            return Arrays.stream(policyStr.split(","))
                         .map(String::trim)
                         .filter(s -> !s.isEmpty())
                         .map(String::toLowerCase)
                         .collect(Collectors.toSet());
        }
        // This is unexpected, as the topic config should include the cleanup.policy even if
        // the topic settings don't override the broker's log.cleanup.policy. But just to be safe.
        log.debug("Found no cleanup.policy for topic '{}'", topic);
        return Collections.emptySet();
    }
Get the cleanup policy for a topic.

@param topic the name of the topic
@return the set of cleanup policies set for the topic; may be empty if the topic does not
        exist or the topic's cleanup policy could not be retrieved
connect/runtime/src/main/java/org/apache/kafka/connect/util/TopicAdmin.java