[GitHub] [samza] mynameborat commented on a change in pull request #1050: SAMZA-2221: Use the KafkaStreamSpec instead of the StreamSpec

[GitBox <gi...@apache.org>]

mynameborat commented on a change in pull request #1050: SAMZA-2221: Use the KafkaStreamSpec instead of the StreamSpec 
URL: https://github.com/apache/samza/pull/1050#discussion_r301285981
 
 

 ##########
 File path: samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaSystemAdmin.java
 ##########
 @@ -461,24 +461,13 @@ public Integer offsetComparator(String offset1, String offset2) {
   @Override
   public boolean createStream(StreamSpec streamSpec) {
     LOG.info("Creating Kafka topic: {} on system: {}", streamSpec.getPhysicalName(), streamSpec.getSystemName());
-    final String REPL_FACTOR = "replication.factor";
 
     KafkaStreamSpec kSpec = toKafkaSpec(streamSpec);
     String topicName = kSpec.getPhysicalName();
 
     // create topic.
     NewTopic newTopic = new NewTopic(topicName, kSpec.getPartitionCount(), (short) kSpec.getReplicationFactor());
-
-    // specify the configs
-    Map<String, String> streamConfig = new HashMap<>(streamSpec.getConfig());
-    // HACK - replication.factor is invalid config for AdminClient.createTopics
-    if (streamConfig.containsKey(REPL_FACTOR)) {
-      String repl = streamConfig.get(REPL_FACTOR);
-      LOG.warn("Configuration {}={} for topic={} is invalid. Using kSpec repl factor {}",
-          REPL_FACTOR, repl, kSpec.getPhysicalName(), kSpec.getReplicationFactor());
-      streamConfig.remove(REPL_FACTOR);
-    }
-    newTopic.configs(new MapConfig(streamConfig));
+    newTopic.configs(KafkaStreamSpec.filterUnsupportedProperties(kSpec.getConfig()));
 
 Review comment:
   It should be sufficient to pass `kSpec.getConfig()` and you don't need apply filter again.
   `toKafkaSpec(...)` takes care of filtering it unsupported properties.

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services