private static Map<String, String> filterUnsupportedProperties(Map<String, String> originalConfig) {
    Map<String, String> filteredConfig = new HashMap<>();
    for (Map.Entry<String, String> entry: originalConfig.entrySet()) {
      // Kafka requires replication factor, but not as a property, so we have to filter it out.
      if (!KafkaConfig.TOPIC_REPLICATION_FACTOR().equals(entry.getKey())) {
        if (LogConfig.configNames().contains(entry.getKey())) {
          filteredConfig.put(entry.getKey(), entry.getValue());
        } else {
          LOG.warn("Property '{}' is not a valid Kafka topic config. It will be ignored.", entry.getKey());
        }
      }
    }
    return filteredConfig;
  }
Filter out properties from the original config that are not supported by Kafka.
For example, we allow users to set replication.factor as a property of the streams
and then parse it out so we can pass it separately as Kafka requires. But Kafka
will also throw if replication.factor is passed as a property on a new topic.

@param originalConfig  The original config to filter
@return                The filtered config
samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaStreamSpec.java