From e058c507a0ec2c1fb0058390677009073586addc Mon Sep 17 00:00:00 2001 From: qiumingming Date: Fri, 30 Jun 2017 18:31:52 +0800 Subject: [PATCH] KYLIN-2700 Should allow user to override Kafka conf in cube level --- .../java/org/apache/kylin/source/kafka/hadoop/KafkaFlatTableJob.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/source-kafka/src/main/java/org/apache/kylin/source/kafka/hadoop/KafkaFlatTableJob.java b/source-kafka/src/main/java/org/apache/kylin/source/kafka/hadoop/KafkaFlatTableJob.java index 11466e5..e98b784 100644 --- a/source-kafka/src/main/java/org/apache/kylin/source/kafka/hadoop/KafkaFlatTableJob.java +++ b/source-kafka/src/main/java/org/apache/kylin/source/kafka/hadoop/KafkaFlatTableJob.java @@ -104,7 +104,7 @@ public class KafkaFlatTableJob extends AbstractHadoopJob { job.getConfiguration().addResource(new Path(jobEngineConfig.getHadoopJobConfFilePath(null))); KafkaConsumerProperties kafkaConsumerProperties = KafkaConsumerProperties.getInstanceFromEnv(); job.getConfiguration().addResource(new Path(kafkaConsumerProperties.getKafkaConsumerHadoopJobConf())); - appendKafkaOverrideProperties(KylinConfig.getInstanceFromEnv(), job.getConfiguration()); + appendKafkaOverrideProperties(cube.getConfig(), job.getConfiguration()); job.getConfiguration().set(CONFIG_KAFKA_BROKERS, brokers); job.getConfiguration().set(CONFIG_KAFKA_TOPIC, topic); job.getConfiguration().set(CONFIG_KAFKA_TIMEOUT, String.valueOf(kafkaConfig.getTimeout())); -- 2.10.1