@SuppressWarnings("unchecked")
    public static List<String> parseTopicsList(Map<String, String> props) {
        List<String> topics = (List<String>) ConfigDef.parseType(TOPICS_CONFIG, props.get(TOPICS_CONFIG), Type.LIST);
        if (topics == null) {
            return Collections.emptyList();
        }
        return topics
                .stream()
                .filter(topic -> !topic.isEmpty())
                .distinct()
                .collect(Collectors.toList());
    }
Perform preflight validation for the sink-specific properties for a connector.

@param props           the configuration for the sink connector
@param validatedConfig any already-known {@link ConfigValue validation results} for the configuration.
                       May be empty, but may not be null. Any configuration errors discovered by this method will
                       be {@link ConfigValue#addErrorMessage(String) added} to a value in this map, adding a new
                       entry if one for the problematic property does not already exist.
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/SinkConnectorConfig.java