KAFKA-10477: Enabling the same behavior of NULL JsonNodeType to MISSING JsonNodeType in JsonConverter.

From 2.10.0 onwards, in jackson lib, ObjectMapper.readTree(input) started to return JsonNode of type MISSING for empty input, as mentioned in the issue: FasterXML/jackson-databind#2211.
Made changes in JsonParser to incorporate this, and treat MISSING JsonNodeType in a similar fashion as that of NULL JsonNodeType.
Added a unit test for this.

Author: shaikzakiriitm

connect/json/src/main/java/org/apache/kafka/connect/json/JsonConverter.java

@@ -729,6 +729,7 @@ private static Object convertToConnect(Schema schema, JsonNode jsonValue) {
         } else {
             switch (jsonValue.getNodeType()) {
                 case NULL:
+                case MISSING:
                     // Special case. With no schema
                     return null;
                 case BOOLEAN:
@@ -751,7 +752,6 @@ private static Object convertToConnect(Schema schema, JsonNode jsonValue) {
                     break;
 
                 case BINARY:
-                case MISSING:
                 case POJO:
                 default:
                     schemaType = null;

connect/json/src/test/java/org/apache/kafka/connect/json/JsonConverterTest.java

@@ -195,6 +195,20 @@ public void nullToConnect() {
         assertEquals(SchemaAndValue.NULL, converted);
     }
 
+    /**
+     * When schemas are disabled, empty data should be decoded to an empty envelope.
+     * This test verifies the case where `schemas.enable` configuration is set to false, and
+     * {@link JsonConverter} converts empty bytes to {@link SchemaAndValue#NULL}.
+     */
+    @Test
+    public void emptyBytesToConnect() {
+        // This characterizes the messages with empty data when Json schemas is disabled
+        Map<String, Boolean> props = Collections.singletonMap("schemas.enable", false);
+        converter.configure(props, true);
+        SchemaAndValue converted = converter.toConnectData(TOPIC, "".getBytes());
+        assertEquals(SchemaAndValue.NULL, converted);
+    }
+
     @Test
     public void nullSchemaPrimitiveToConnect() {
         SchemaAndValue converted = converter.toConnectData(TOPIC, "{ \"schema\": null, \"payload\": null }".getBytes());