Fix broken KafkaEmitterConfig parsing (#5201)

* Fix broken KafkaEmitterConfig parsing

This was a regression introduced in
https://github.com/druid-io/druid/pull/4722

KafkaEmitterConfig property names have dot(.) in the name of properties
and JsonConfigurator behavior was changed to not support that.
Added a test and fixed parsing of properties that have dot(.) in
property names

* Fix test failure

api/src/main/java/io/druid/guice/JsonConfigurator.java

@@ -93,7 +93,6 @@ public class JsonConfigurator
           log.info(e, "Unable to parse [%s]=[%s] as a json object, using as is.", prop, propValue);
           value = propValue;
         }
-
         hieraricalPutValue(propertyPrefix, prop, prop.substring(propertyBase.length()), value, jsonMap);
       }
     }
@@ -175,8 +174,11 @@ public class JsonConfigurator
   )
   {
     int dotIndex = property.indexOf('.');
+    // Always put property with name even if it is of form a.b. This will make sure the property is available for classes
+    // where JsonProperty names are of the form a.b
+    // Note:- this will cause more than required properties to be present in the jsonMap.
+    targetMap.put(property, value);
     if (dotIndex < 0) {
-      targetMap.put(property, value);
       return;
     }
     if (dotIndex == 0) {

api/src/test/java/io/druid/guice/JsonConfiguratorTest.java

@@ -94,10 +94,13 @@ public class JsonConfiguratorTest
   public void testTest()
   {
     Assert.assertEquals(
-        new MappableObject("p1", ImmutableList.<String>of("p2")),
-        new MappableObject("p1", ImmutableList.<String>of("p2"))
+        new MappableObject("p1", ImmutableList.<String>of("p2"), "p2"),
+        new MappableObject("p1", ImmutableList.<String>of("p2"), "p2")
+    );
+    Assert.assertEquals(
+        new MappableObject("p1", null, null),
+        new MappableObject("p1", ImmutableList.<String>of(), null)
     );
-    Assert.assertEquals(new MappableObject("p1", null), new MappableObject("p1", ImmutableList.<String>of()));
   }
 
   @Test
@@ -140,6 +143,19 @@ public class JsonConfiguratorTest
     Assert.assertEquals("testing \"prop1\"", obj.prop1);
     Assert.assertEquals(ImmutableList.of(), obj.prop1List);
   }
+
+  @Test
+  public void testPropertyWithDot()
+  {
+    final JsonConfigurator configurator = new JsonConfigurator(mapper, validator);
+    properties.setProperty(PROP_PREFIX + "prop2.prop.2", "testing");
+    properties.setProperty(PROP_PREFIX + "prop1", "prop1");
+    final MappableObject obj = configurator.configurate(properties, PROP_PREFIX, MappableObject.class);
+    Assert.assertEquals("testing", obj.prop2);
+    Assert.assertEquals(ImmutableList.of(), obj.prop1List);
+    Assert.assertEquals("prop1", obj.prop1);
+
+  }
 }
 
 class MappableObject
@@ -148,15 +164,19 @@ class MappableObject
   final String prop1;
   @JsonProperty("prop1List")
   final List<String> prop1List;
+  @JsonProperty("prop2.prop.2")
+  final String prop2;
 
   @JsonCreator
   protected MappableObject(
       @JsonProperty("prop1") final String prop1,
-      @JsonProperty("prop1List") final List<String> prop1List
+      @JsonProperty("prop1List") final List<String> prop1List,
+      @JsonProperty("prop2.prop.2") final String prop2
   )
   {
     this.prop1 = prop1;
     this.prop1List = prop1List == null ? ImmutableList.<String>of() : prop1List;
+    this.prop2 = prop2;
   }
 
 
@@ -172,6 +192,12 @@ class MappableObject
     return prop1;
   }
 
+  @JsonProperty
+  public String getProp2()
+  {
+    return prop2;
+  }
+
   @Override
   public boolean equals(Object o)
   {

server/src/test/java/io/druid/server/emitter/EmitterModuleTest.java

@@ -31,6 +31,7 @@ import io.druid.guice.JsonConfigurator;
 import io.druid.guice.LazySingleton;
 import io.druid.guice.LifecycleModule;
 import io.druid.guice.ServerModule;
+import io.druid.jackson.JacksonModule;
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -67,6 +68,7 @@ public class EmitterModuleTest
             new DruidGuiceExtensions(),
             new LifecycleModule(),
             new ServerModule(),
+            new JacksonModule(),
             new Module()
             {
               @Override