Merge branch '4.0.x' into 4.1.x

Author: rhauch

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

@@ -197,10 +197,12 @@ private static JsonNode inferPrimitive(String type, Object defaultValue) {
           break;
         case ElasticsearchSinkConnectorConstants.STRING_TYPE:
         case ElasticsearchSinkConnectorConstants.TEXT_TYPE:
-          defaultValueNode = JsonNodeFactory.instance.textNode((String) defaultValue);
-          break;
         case ElasticsearchSinkConnectorConstants.BINARY_TYPE:
-          defaultValueNode = JsonNodeFactory.instance.binaryNode(bytes(defaultValue));
+          // IGNORE default values for text and binary types as this is not supported by ES side.
+          // see https://www.elastic.co/guide/en/elasticsearch/reference/current/text.html
+          // https://www.elastic.co/guide/en/elasticsearch/reference/current/binary.html
+          // for more details.
+          //defaultValueNode = null;
           break;
         case ElasticsearchSinkConnectorConstants.BOOLEAN_TYPE:
           defaultValueNode = JsonNodeFactory.instance.booleanNode((boolean) defaultValue);

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

@@ -16,6 +16,7 @@
 
 package io.confluent.connect.elasticsearch;
 
+import com.fasterxml.jackson.databind.JsonNode;
 import com.google.gson.JsonObject;
 
 import org.apache.kafka.connect.data.Date;
@@ -51,6 +52,29 @@ public void testMapping() throws Exception {
     verifyMapping(schema, mapping);
   }
 
+  @Test
+  public void testInferMapping() throws Exception {
+
+    Schema stringSchema = SchemaBuilder
+        .struct()
+        .name("record")
+        .field("foo", SchemaBuilder.string().defaultValue("0").build())
+        .build();
+    JsonNode stringMapping = Mapping.inferMapping(client, stringSchema);
+
+    assertNull(stringMapping.get("properties").get("foo").get("null_value"));
+
+    Schema intSchema =SchemaBuilder
+        .struct()
+        .name("record")
+        .field("foo", SchemaBuilder.int32().defaultValue(0).build())
+        .build();
+
+    JsonNode intMapping = Mapping.inferMapping(client, intSchema);
+    assertNotNull(intMapping.get("properties").get("foo").get("null_value"));
+    assertEquals(0, intMapping.get("properties").get("foo").get("null_value").asInt());
+  }
+
   protected Schema createSchema() {
     Schema structSchema = createInnerSchema();
     return SchemaBuilder.struct().name("record")
@@ -96,7 +120,6 @@ private Schema createInnerSchema() {
   @SuppressWarnings("unchecked")
   private void verifyMapping(Schema schema, JsonObject mapping) throws Exception {
     String schemaName = schema.name();
-
     Object type = mapping.get("type");
     if (schemaName != null) {
       switch (schemaName) {