Merge branch '4.1.x' into 5.0.x

Author: rhauch

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

@@ -201,10 +201,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

@@ -19,6 +19,7 @@
 import com.fasterxml.jackson.databind.node.NumericNode;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.fasterxml.jackson.databind.node.TextNode;
+import com.fasterxml.jackson.databind.JsonNode;
 import com.google.gson.JsonObject;
 
 import org.apache.kafka.connect.data.Date;
@@ -35,7 +36,6 @@
 import static io.confluent.connect.elasticsearch.DataConverter.BehaviorOnNullValues;
 import static io.confluent.connect.elasticsearch.ElasticsearchSinkConnectorConstants.KEYWORD_TYPE;
 import static io.confluent.connect.elasticsearch.ElasticsearchSinkConnectorConstants.TEXT_TYPE;
-import static org.junit.Assert.assertEquals;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
@@ -82,6 +82,29 @@ public void testStringMappingForES6() throws Exception {
     assertEquals(256, ignoreAbove.asInt());
   }
 
+  @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")
@@ -127,7 +150,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) {