Repository: embulk/embulk
Issue: Plugin's Task causes JsonMappingException when the Task extends TimestampFormatter.Task with @Config("default_timezone")

When a plugin has its `Task` interface which extends core's `Task` interface with `default` methods for compatibility, it causes `JsonMappingException` such as:

```
com.fasterxml.jackson.databind.JsonMappingException: Multiple entries with same key: default_timezone=org.embulk.config.TaskSerDe$TaskDeserializer$FieldEntry@2548fc01 and default_timezone=org.embulk.config.TaskSerDe$TaskDeserializer$FieldEntry@4331be30
```

embulk-filter-calcite is the case.
https://github.com/muga/embulk-filter-calcite/blob/v0.1.3/src/main/java/org/embulk/filter/calcite/CalciteFilterPlugin.java#L246-L248

The direct cause is that the extended `interface` has two getter methods with the same `@Config("default_timezone")`.

Options to fix it:

1. If the parent's method has `default` implementation, and it does not have `@Config`, the overridden method behaves like it does not have `@Config`. In calcite's case, `getDefaultTimeZone` behaves like it does not have `@Config`.
2. Multiple getters with the same `@Config` configuration names are allowed.

Changed files:
embulk-core/src/test/java/org/embulk/config/TestConfigSource.java, embulk-core/src/main/java/org/embulk/config/TaskSerDe.java, embulk-core/src/main/java/org/embulk/config/TaskInvocationHandler.java

diff --git a/embulk-core/src/main/java/org/embulk/config/TaskInvocationHandler.java b/embulk-core/src/main/java/org/embulk/config/TaskInvocationHandler.java
index 53fc4d35..a652c44a 100644
--- a/embulk-core/src/main/java/org/embulk/config/TaskInvocationHandler.java
+++ b/embulk-core/src/main/java/org/embulk/config/TaskInvocationHandler.java
@@ -1,6 +1,7 @@
 package org.embulk.config;
 
-import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableMultimap;
+import com.google.common.collect.Multimap;
 import java.lang.invoke.MethodHandles;
 import java.lang.reflect.Constructor;
 import java.lang.reflect.InvocationHandler;
@@ -24,10 +25,12 @@ class TaskInvocationHandler implements InvocationHandler {
     }
 
     /**
-     * fieldName = Method of the getter
+     * Returns a Multimap from fieldName Strings to their getter Methods.
+     *
+     * It expects to be called only from TaskSerDe. Multimap is used inside org.embulk.config.
      */
-    public static Map<String, Method> fieldGetters(Class<?> iface) {
-        ImmutableMap.Builder<String, Method> builder = ImmutableMap.builder();
+    static Multimap<String, Method> fieldGetters(Class<?> iface) {
+        ImmutableMultimap.Builder<String, Method> builder = ImmutableMultimap.builder();
         for (Method method : iface.getMethods()) {
             String methodName = method.getName();
             String fieldName = getterFieldNameOrNull(methodName);
diff --git a/embulk-core/src/main/java/org/embulk/config/TaskSerDe.java b/embulk-core/src/main/java/org/embulk/config/TaskSerDe.java
index 28df6199..677a6486 100644
--- a/embulk-core/src/main/java/org/embulk/config/TaskSerDe.java
+++ b/embulk-core/src/main/java/org/embulk/config/TaskSerDe.java
@@ -10,6 +10,7 @@ import com.fasterxml.jackson.databind.DeserializationContext;
 import com.fasterxml.jackson.databind.JavaType;
 import com.fasterxml.jackson.databind.JsonDeserializer;
 import com.fasterxml.jackson.databind.JsonMappingException;
+import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.JsonSerializer;
 import com.fasterxml.jackson.databind.Module;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -17,14 +18,16 @@ import com.fasterxml.jackson.databind.SerializerProvider;
 import com.fasterxml.jackson.databind.deser.Deserializers;
 import com.fasterxml.jackson.databind.module.SimpleModule;
 import com.google.common.base.Optional;
+import com.google.common.collect.HashMultimap;
 import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableMultimap;
 import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Multimap;
 import java.io.IOException;
 import java.lang.reflect.Method;
 import java.lang.reflect.Proxy;
 import java.lang.reflect.Type;
-import java.util.HashMap;
+import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
@@ -66,7 +69,7 @@ class TaskSerDe {
         private final ObjectMapper nestedObjectMapper;
         private final ModelManager model;
         private final Class<?> iface;
-        private final Map<String, FieldEntry> mappings;
+        private final Multimap<String, FieldEntry> mappings;
         private final List<InjectEntry> injects;
 
         public TaskDeserializer(ObjectMapper nestedObjectMapper, ModelManager model, Class<T> iface) {
@@ -77,9 +80,9 @@ class TaskSerDe {
             this.injects = injectEntries(iface);
         }
 
-        protected Map<String, FieldEntry> getterMappings(Class<?> iface) {
-            ImmutableMap.Builder<String, FieldEntry> builder = ImmutableMap.builder();
-            for (Map.Entry<String, Method> getter : TaskInvocationHandler.fieldGetters(iface).entrySet()) {
+        protected Multimap<String, FieldEntry> getterMappings(Class<?> iface) {
+            ImmutableMultimap.Builder<String, FieldEntry> builder = ImmutableMultimap.builder();
+            for (Map.Entry<String, Method> getter : TaskInvocationHandler.fieldGetters(iface).entries()) {
                 Method getterMethod = getter.getValue();
                 String fieldName = getter.getKey();
 
@@ -103,7 +106,7 @@ class TaskSerDe {
 
         protected List<InjectEntry> injectEntries(Class<?> iface) {
             ImmutableList.Builder<InjectEntry> builder = ImmutableList.builder();
-            for (Map.Entry<String, Method> getter : TaskInvocationHandler.fieldGetters(iface).entrySet()) {
+            for (Map.Entry<String, Method> getter : TaskInvocationHandler.fieldGetters(iface).entries()) {
                 Method getterMethod = getter.getValue();
                 String fieldName = getter.getKey();
                 ConfigInject inject = getterMethod.getAnnotation(ConfigInject.class);
@@ -127,7 +130,7 @@ class TaskSerDe {
         @SuppressWarnings("unchecked")
         public T deserialize(JsonParser jp, DeserializationContext ctxt) throws IOException {
             Map<String, Object> objects = new ConcurrentHashMap<String, Object>();
-            HashMap<String, FieldEntry> unusedMappings = new HashMap<>(mappings);
+            HashMultimap<String, FieldEntry> unusedMappings = HashMultimap.<String, FieldEntry>create(mappings);
 
             String key;
             JsonToken current = jp.getCurrentToken();
@@ -140,21 +143,31 @@ class TaskSerDe {
 
             for (; key != null; key = jp.nextFieldName()) {
                 JsonToken t = jp.nextToken(); // to get to value
-                FieldEntry field = mappings.get(key);
-                if (field == null) {
+                final Collection<FieldEntry> fields = mappings.get(key);
+                if (fields.isEmpty()) {
                     jp.skipChildren();
                 } else {
-                    Object value = nestedObjectMapper.readValue(jp, new GenericTypeReference(field.getType()));
-                    if (value == null) {
-                        throw new JsonMappingException("Setting null to a task field is not allowed. Use Optional<T> (com.google.common.base.Optional) to represent null.");
+                    final JsonNode children = nestedObjectMapper.readValue(jp, JsonNode.class);
+                    for (final FieldEntry field : fields) {
+                        final Object value = nestedObjectMapper.convertValue(children, new GenericTypeReference(field.getType()));
+                        if (value == null) {
+                            throw new JsonMappingException("Setting null to a task field is not allowed. Use Optional<T> (com.google.common.base.Optional) to represent null.");
+                        }
+                        objects.put(field.getName(), value);
+                        if (!unusedMappings.remove(key, field)) {
+                            throw new JsonMappingException(String.format(
+                                    "FATAL: Expected to be a bug in Embulk. Mapping \\"%s: (%s) %s\\" might have already been processed, or not in %s.",
+                                    key,
+                                    field.getType().toString(),
+                                    field.getName(),
+                                    this.iface.toString()));
+                        }
                     }
-                    objects.put(field.getName(), value);
-                    unusedMappings.remove(key);
                 }
             }
 
             // set default values
-            for (Map.Entry<String, FieldEntry> unused : unusedMappings.entrySet()) {
+            for (Map.Entry<String, FieldEntry> unused : unusedMappings.entries()) {
                 FieldEntry field = unused.getValue();
                 if (field.getDefaultJsonString().isPresent()) {
                     Object value = nestedObjectMapper.readValue(field.getDefaultJsonString().get(), new GenericTypeReference(field.getType()));
diff --git a/embulk-core/src/test/java/org/embulk/config/TestConfigSource.java b/embulk-core/src/test/java/org/embulk/config/TestConfigSource.java
index 73c04563..62a406a7 100644
--- a/embulk-core/src/test/java/org/embulk/config/TestConfigSource.java
+++ b/embulk-core/src/test/java/org/embulk/config/TestConfigSource.java
@@ -4,6 +4,7 @@ import static org.junit.Assert.assertEquals;
 
 import org.embulk.EmbulkTestRuntime;
 import org.embulk.spi.Exec;
+import org.embulk.spi.time.TimestampParser;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
@@ -36,6 +37,28 @@ public class TestConfigSource {
         public String getString();
     }
 
+    private static interface DuplicationParent extends Task {
+        @Config("duplicated_number")
+        public int getInteger();
+    }
+
+    private static interface Duplicated extends DuplicationParent {
+        @Config("duplicated_number")
+        public String getString();
+
+        @Config("duplicated_number")
+        public double getDouble();
+    }
+
+    private static interface DuplicatedDateTimeZone extends Task, TimestampParser.Task {
+        @Config("default_timezone")
+        @ConfigDefault("\\"America/Los_Angeles\\"")
+        public org.joda.time.DateTimeZone getDefaultTimeZone();
+
+        @Config("dummy_value")
+        public String getDummyValue();
+    }
+
     @Test
     public void testSetGet() {
         config.set("boolean", true);
@@ -67,6 +90,43 @@ public class TestConfigSource {
         assertEquals("sf", task.getString());
     }
 
+    @Test
+    public void testDuplicatedConfigName() {
+        config.set("duplicated_number", "1034");
+
+        Duplicated task = config.loadConfig(Duplicated.class);
+        assertEquals(1034, task.getInteger());
+        assertEquals("1034", task.getString());
+        assertEquals(1034.0, task.getDouble(), 0.000001);
+    }
+
+    @Test
+    public void testDuplicatedDateTimeZone() {
+        config.set("default_timezone", "Asia/Tokyo");
+        config.set("default_timestamp_format", "%Y");
+        config.set("dummy_value", "foobar");
+
+        DuplicatedDateTimeZone task = config.loadConfig(DuplicatedDateTimeZone.class);
+        assertEquals("Asia/Tokyo", task.getDefaultTimeZoneId());
+        assertEquals(org.joda.time.DateTimeZone.forID("Asia/Tokyo"), task.getDefaultTimeZone());
+        assertEquals("%Y", task.getDefaultTimestampFormat());
+        assertEquals("1970-01-01", task.getDefaultDate());
+        assertEquals("foobar", task.getDummyValue());
+    }
+
+    @Test
+    public void testDuplicatedDateTimeZoneWithDefault() {
+        config.set("default_timestamp_format", "%Y");
+        config.set("dummy_value", "foobar");
+
+        DuplicatedDateTimeZone task = config.loadConfig(DuplicatedDateTimeZone.class);
+        assertEquals("UTC", task.getDefaultTimeZoneId());
+        assertEquals(org.joda.time.DateTimeZone.forID("America/Los_Angeles"), task.getDefaultTimeZone());
+        assertEquals("%Y", task.getDefaultTimestampFormat());
+        assertEquals("1970-01-01", task.getDefaultDate());
+        assertEquals("foobar", task.getDummyValue());
+    }
+
     private static interface ValidateFields extends Task {
         @Config("valid")
         public String getValid();