From 324e3db7bcd65acf851fe93f2135d02c80c000ac Mon Sep 17 00:00:00 2001 From: Yuewen Wang Date: Thu, 27 Jul 2017 20:17:39 +0800 Subject: [PATCH 1/3] Fix a NPE using kafka emitter extension --- .../io/druid/emitter/kafka/KafkaEmitterConfig.java | 9 ++++----- .../emitter/kafka/KafkaEmitterConfigTest.java | 14 ++++++++++++++ 2 files changed, 18 insertions(+), 5 deletions(-) diff --git a/extensions-contrib/kafka-emitter/src/main/java/io/druid/emitter/kafka/KafkaEmitterConfig.java b/extensions-contrib/kafka-emitter/src/main/java/io/druid/emitter/kafka/KafkaEmitterConfig.java index b5ca3ce096de..c66cca93baf2 100644 --- a/extensions-contrib/kafka-emitter/src/main/java/io/druid/emitter/kafka/KafkaEmitterConfig.java +++ b/extensions-contrib/kafka-emitter/src/main/java/io/druid/emitter/kafka/KafkaEmitterConfig.java @@ -22,6 +22,7 @@ import com.fasterxml.jackson.annotation.JsonCreator; import com.fasterxml.jackson.annotation.JsonProperty; import com.google.common.base.Preconditions; +import com.google.common.collect.ImmutableMap; import org.apache.kafka.clients.producer.ProducerConfig; import java.util.Map; @@ -53,7 +54,7 @@ public KafkaEmitterConfig( this.metricTopic = Preconditions.checkNotNull(metricTopic, "metric.topic can not be null"); this.alertTopic = Preconditions.checkNotNull(alertTopic, "alert.topic can not be null"); this.clusterName = clusterName; - this.kafkaProducerConfig = kafkaProducerConfig; + this.kafkaProducerConfig = kafkaProducerConfig == null? ImmutableMap.of() : kafkaProducerConfig; } @JsonProperty @@ -110,9 +111,7 @@ public boolean equals(Object o) if (getClusterName() != null ? !getClusterName().equals(that.getClusterName()) : that.getClusterName() != null) { return false; } - return getKafkaProducerConfig() != null - ? getKafkaProducerConfig().equals(that.getKafkaProducerConfig()) - : that.getKafkaProducerConfig() == null; + return getKafkaProducerConfig().equals(that.getKafkaProducerConfig()); } @Override @@ -122,7 +121,7 @@ public int hashCode() result = 31 * result + getMetricTopic().hashCode(); result = 31 * result + getAlertTopic().hashCode(); result = 31 * result + (getClusterName() != null ? getClusterName().hashCode() : 0); - result = 31 * result + (getKafkaProducerConfig() != null ? getKafkaProducerConfig().hashCode() : 0); + result = 31 * result + getKafkaProducerConfig().hashCode(); return result; } diff --git a/extensions-contrib/kafka-emitter/src/test/java/io/druid/emitter/kafka/KafkaEmitterConfigTest.java b/extensions-contrib/kafka-emitter/src/test/java/io/druid/emitter/kafka/KafkaEmitterConfigTest.java index 37a8c1133486..b9ab4e58a6c4 100644 --- a/extensions-contrib/kafka-emitter/src/test/java/io/druid/emitter/kafka/KafkaEmitterConfigTest.java +++ b/extensions-contrib/kafka-emitter/src/test/java/io/druid/emitter/kafka/KafkaEmitterConfigTest.java @@ -52,4 +52,18 @@ public void testSerDeserKafkaEmitterConfig() throws IOException .readValue(kafkaEmitterConfigString); Assert.assertEquals(kafkaEmitterConfigExpected, kafkaEmitterConfig); } + + @Test + public void testSerDeNotRequiredKafkaProducerConfig() throws IOException + { + KafkaEmitterConfig kafkaEmitterConfig = new KafkaEmitterConfig("localhost:9092", "metricTest", + "alertTest", "clusterNameTest", + null + ); + try { + KafkaEmitter emitter = new KafkaEmitter(kafkaEmitterConfig, mapper); + } catch (NullPointerException e) { + Assert.fail(); + } + } } From 86aaaff05f658a6575cf1d5c55eafcf65be68167 Mon Sep 17 00:00:00 2001 From: Yuewen Wang Date: Fri, 28 Jul 2017 10:23:04 +0800 Subject: [PATCH 2/3] fix format --- .../java/io/druid/emitter/kafka/KafkaEmitterConfigTest.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/extensions-contrib/kafka-emitter/src/test/java/io/druid/emitter/kafka/KafkaEmitterConfigTest.java b/extensions-contrib/kafka-emitter/src/test/java/io/druid/emitter/kafka/KafkaEmitterConfigTest.java index b9ab4e58a6c4..3b91706fbf02 100644 --- a/extensions-contrib/kafka-emitter/src/test/java/io/druid/emitter/kafka/KafkaEmitterConfigTest.java +++ b/extensions-contrib/kafka-emitter/src/test/java/io/druid/emitter/kafka/KafkaEmitterConfigTest.java @@ -62,7 +62,8 @@ public void testSerDeNotRequiredKafkaProducerConfig() throws IOException ); try { KafkaEmitter emitter = new KafkaEmitter(kafkaEmitterConfig, mapper); - } catch (NullPointerException e) { + } + catch (NullPointerException e) { Assert.fail(); } } From ea6dbd82f07e119d528c2cea1c86663518eb8d62 Mon Sep 17 00:00:00 2001 From: Yuewen Wang Date: Mon, 31 Jul 2017 10:14:01 +0800 Subject: [PATCH 3/3] Add @Nullable annotation on kafkaProducerConfig --- .../main/java/io/druid/emitter/kafka/KafkaEmitterConfig.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/extensions-contrib/kafka-emitter/src/main/java/io/druid/emitter/kafka/KafkaEmitterConfig.java b/extensions-contrib/kafka-emitter/src/main/java/io/druid/emitter/kafka/KafkaEmitterConfig.java index c66cca93baf2..2b73f59b6dcc 100644 --- a/extensions-contrib/kafka-emitter/src/main/java/io/druid/emitter/kafka/KafkaEmitterConfig.java +++ b/extensions-contrib/kafka-emitter/src/main/java/io/druid/emitter/kafka/KafkaEmitterConfig.java @@ -25,6 +25,7 @@ import com.google.common.collect.ImmutableMap; import org.apache.kafka.clients.producer.ProducerConfig; +import javax.annotation.Nullable; import java.util.Map; public class KafkaEmitterConfig @@ -47,7 +48,7 @@ public KafkaEmitterConfig( @JsonProperty("metric.topic") String metricTopic, @JsonProperty("alert.topic") String alertTopic, @JsonProperty("clusterName") String clusterName, - @JsonProperty("producer.config") Map kafkaProducerConfig + @JsonProperty("producer.config") @Nullable Map kafkaProducerConfig ) { this.bootstrapServers = Preconditions.checkNotNull(bootstrapServers, "bootstrap.servers can not be null");