Merge branch '4.1.x' into 5.0.x

Author: ConfluentJenkins

src/main/java/io/confluent/connect/elasticsearch/Mapping.java

@@ -212,7 +212,8 @@ private static JsonNode inferPrimitive(String type, Object defaultValue) {
           defaultValueNode = JsonNodeFactory.instance.booleanNode((boolean) defaultValue);
           break;
         case ElasticsearchSinkConnectorConstants.DATE_TYPE:
-          defaultValueNode = JsonNodeFactory.instance.numberNode((long) defaultValue);
+          long value = ((java.util.Date) defaultValue).getTime();
+          defaultValueNode = JsonNodeFactory.instance.numberNode(value);
           break;
         default:
           throw new DataException("Invalid primitive type.");

src/test/java/io/confluent/connect/elasticsearch/MappingTest.java

@@ -105,6 +105,24 @@ public void testInferMapping() throws Exception {
     assertEquals(0, intMapping.get("properties").get("foo").get("null_value").asInt());
   }
 
+  @Test
+  public void testInferMappingDefaultDate()  {
+    java.util.Date expected = new java.util.Date();
+
+    Schema dateSchema = SchemaBuilder
+        .struct()
+        .name("record")
+        .field("foo", Date.builder().defaultValue(expected).build())
+        .build();
+
+    JsonNode dateMapping = Mapping.inferMapping(client, dateSchema);
+    assertNotNull(dateMapping.get("properties").get("foo").get("null_value"));
+    assertEquals(
+        expected.getTime(),
+        dateMapping.get("properties").get("foo").get("null_value").asLong()
+    );
+  }
+
   protected Schema createSchema() {
     Schema structSchema = createInnerSchema();
     return SchemaBuilder.struct().name("record")