From cb9641875b8b9783f06f5c1169231d36d75ae76f Mon Sep 17 00:00:00 2001
From: javalife0312 <javalife0312@126.com>
Date: Fri, 12 Apr 2019 15:09:11 +0800
Subject: [PATCH] KYLIN-3932 KafkaConfigOverride to take effect

---
 .../org/apache/kylin/stream/source/kafka/KafkaSource.java  | 7 ++++++-
 1 file changed, 6 insertions(+), 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
index 1f61f432456..6e46c687610 100644
--- 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) {