KYLIN-3932 KafkaConfigOverride to take effect

apache · Apr 12, 2019 · cb96418 · cb96418
1 parent e13f101
commit cb96418
Showing 1 changed file with 6 additions and 1 deletion.
diff --git a/stream-source-kafka/src/main/java/org/apache/kylin/stream/source/kafka/KafkaSource.java b/stream-source-kafka/src/main/java/org/apache/kylin/stream/source/kafka/KafkaSource.java
@@ -245,7 +245,12 @@ private boolean isEmptyPosition(KafkaPosition kafkaPosition) {
     }
 
     public static Map<String, Object> getKafkaConf(Map<String, String> sourceProperties, KylinConfig kylinConfig) {
-        return getKafkaConf(sourceProperties);
+        Map<String, String> kafkaConfigOverride = kylinConfig.getKafkaConfigOverride();
+        Map<String, Object> kafkaConf = getKafkaConf(sourceProperties);
+        kafkaConf.putAll(kafkaConfigOverride);
+
+        return kafkaConf;
+        //return getKafkaConf(sourceProperties);
     }
 
     public static Map<String, Object> getKafkaConf(Map<String, String> sourceProperties) {