From 3c457629f44cf01417178b5b545347f98bc5969f Mon Sep 17 00:00:00 2001
From: Harini Rajendran <hrajendran@confluent.io>
Date: Mon, 31 Oct 2022 18:40:49 -0500
Subject: [PATCH] wip

---
 .../util/emitter/service/SegmentMetadataEvent.java |  3 ---
 .../emitter/service/SegmentMetadataEventTest.java  | 14 +++++++-------
 extensions-contrib/kafka-emitter/pom.xml           |  2 ++
 .../apache/druid/emitter/kafka/KafkaEmitter.java   |  2 +-
 .../druid/emitter/kafka/KafkaEmitterConfig.java    |  3 +--
 5 files changed, 11 insertions(+), 13 deletions(-)

diff --git a/core/src/main/java/org/apache/druid/java/util/emitter/service/SegmentMetadataEvent.java b/core/src/main/java/org/apache/druid/java/util/emitter/service/SegmentMetadataEvent.java
index 0ba629da2afc..01577c47f6b3 100644
--- a/core/src/main/java/org/apache/druid/java/util/emitter/service/SegmentMetadataEvent.java
+++ b/core/src/main/java/org/apache/druid/java/util/emitter/service/SegmentMetadataEvent.java
@@ -20,13 +20,10 @@
 package org.apache.druid.java.util.emitter.service;
 
 import com.fasterxml.jackson.annotation.JsonValue;
-import com.google.common.collect.ImmutableMap;
 import org.apache.druid.java.util.emitter.core.Event;
 import org.apache.druid.java.util.emitter.core.EventMap;
 import org.joda.time.DateTime;
 
-import java.util.Map;
-
 public class SegmentMetadataEvent implements Event
 {
   public static final String FEED = "feed";
diff --git a/core/src/test/java/org/apache/druid/java/util/emitter/service/SegmentMetadataEventTest.java b/core/src/test/java/org/apache/druid/java/util/emitter/service/SegmentMetadataEventTest.java
index 3075970259e8..83a4fcba7dc5 100644
--- a/core/src/test/java/org/apache/druid/java/util/emitter/service/SegmentMetadataEventTest.java
+++ b/core/src/test/java/org/apache/druid/java/util/emitter/service/SegmentMetadataEventTest.java
@@ -20,7 +20,7 @@
 package org.apache.druid.java.util.emitter.service;
 
 import com.google.common.collect.ImmutableMap;
-import org.joda.time.DateTime;
+import org.apache.druid.java.util.common.DateTimes;
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -31,9 +31,9 @@ public void testBasicEvent()
   {
     SegmentMetadataEvent event = new SegmentMetadataEvent(
         "dummy_datasource",
-        DateTime.parse("2001-01-01T00:00:00.000Z"),
-        DateTime.parse("2001-01-02T00:00:00.000Z"),
-        DateTime.parse("2001-01-03T00:00:00.000Z"),
+        DateTimes.of("2001-01-01T00:00:00.000Z"),
+        DateTimes.of("2001-01-02T00:00:00.000Z"),
+        DateTimes.of("2001-01-03T00:00:00.000Z"),
         "dummy_version",
         true
     );
@@ -42,9 +42,9 @@ public void testBasicEvent()
         ImmutableMap.<String, Object>builder()
             .put(SegmentMetadataEvent.FEED, "segment_metadata")
             .put(SegmentMetadataEvent.DATASOURCE, "dummy_datasource")
-            .put(SegmentMetadataEvent.CREATED_TIME, DateTime.parse("2001-01-01T00:00:00.000Z"))
-            .put(SegmentMetadataEvent.START_TIME, DateTime.parse("2001-01-02T00:00:00.000Z"))
-            .put(SegmentMetadataEvent.END_TIME, DateTime.parse("2001-01-03T00:00:00.000Z"))
+            .put(SegmentMetadataEvent.CREATED_TIME, DateTimes.of("2001-01-01T00:00:00.000Z"))
+            .put(SegmentMetadataEvent.START_TIME, DateTimes.of("2001-01-02T00:00:00.000Z"))
+            .put(SegmentMetadataEvent.END_TIME, DateTimes.of("2001-01-03T00:00:00.000Z"))
             .put(SegmentMetadataEvent.VERSION, "dummy_version")
             .put(SegmentMetadataEvent.IS_COMPACTED, true)
             .build(),
diff --git a/extensions-contrib/kafka-emitter/pom.xml b/extensions-contrib/kafka-emitter/pom.xml
index 21206500cff9..1cefc780c4cc 100644
--- a/extensions-contrib/kafka-emitter/pom.xml
+++ b/extensions-contrib/kafka-emitter/pom.xml
@@ -126,6 +126,7 @@
       <plugin>
         <groupId>kr.motd.maven</groupId>
         <artifactId>os-maven-plugin</artifactId>
+        <version>1.7.0</version>
         <executions>
           <execution>
             <phase>initialize</phase>
@@ -138,6 +139,7 @@
       <plugin>
         <groupId>org.xolstice.maven.plugins</groupId>
         <artifactId>protobuf-maven-plugin</artifactId>
+        <version>0.6.1</version>
         <executions>
           <execution>
             <goals>
diff --git a/extensions-contrib/kafka-emitter/src/main/java/org/apache/druid/emitter/kafka/KafkaEmitter.java b/extensions-contrib/kafka-emitter/src/main/java/org/apache/druid/emitter/kafka/KafkaEmitter.java
index 6a4a0c1c1dd1..efcad284deb4 100644
--- a/extensions-contrib/kafka-emitter/src/main/java/org/apache/druid/emitter/kafka/KafkaEmitter.java
+++ b/extensions-contrib/kafka-emitter/src/main/java/org/apache/druid/emitter/kafka/KafkaEmitter.java
@@ -218,7 +218,7 @@ public void emit(final Event event)
             requestLost.incrementAndGet();
           }
         } else if (event instanceof SegmentMetadataEvent) {
-          if (!eventTypes.contains(EventType.SEGMENTMETADATA) ) {
+          if (!eventTypes.contains(EventType.SEGMENTMETADATA)) {
             segmentMetadataLost.incrementAndGet();
           } else {
             switch (config.getSegmentMetadataTopicFormat()) {
diff --git a/extensions-contrib/kafka-emitter/src/main/java/org/apache/druid/emitter/kafka/KafkaEmitterConfig.java b/extensions-contrib/kafka-emitter/src/main/java/org/apache/druid/emitter/kafka/KafkaEmitterConfig.java
index 358428cf5301..01a4e39edefc 100644
--- a/extensions-contrib/kafka-emitter/src/main/java/org/apache/druid/emitter/kafka/KafkaEmitterConfig.java
+++ b/extensions-contrib/kafka-emitter/src/main/java/org/apache/druid/emitter/kafka/KafkaEmitterConfig.java
@@ -29,7 +29,6 @@
 import org.apache.kafka.clients.producer.ProducerConfig;
 
 import javax.annotation.Nullable;
-import java.util.Arrays;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
@@ -41,7 +40,7 @@ public enum EventType
     METRICS,
     ALERTS,
     REQUESTS,
-    SEGMENTMETADATA{
+    SEGMENTMETADATA {
       @Override
       public String toString()
       {