feat: addressing review comments

Contributes to: event-integration/eventstreams-planning/12766
KIP Link:  https://cwiki.apache.org/confluence/display/KAFKA/KIP-1054%3A+Support+external+schemas+in+JSONConverte

Signed-off-by: Priyanka K U <[email protected]>

Author: Priyanka K U

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

@@ -36,6 +36,8 @@
 
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
 
 import java.io.File;
 import java.io.IOException;
@@ -980,37 +982,52 @@ public void testVersionRetrievedFromAppInfoParser() {
 
     @Test
     public void testSchemaContentIsNull() {
-        converter.configure(Collections.singletonMap(JsonConverterConfig.SCHEMA_CONTENT_CONFIG, null), false);
+        converter.configure(Map.of(JsonConverterConfig.SCHEMA_CONTENT_CONFIG, null), false);
         assertEquals(new SchemaAndValue(Schema.STRING_SCHEMA, "foo-bar-baz"), converter.toConnectData(TOPIC, "{ \"schema\": { \"type\": \"string\" }, \"payload\": \"foo-bar-baz\" }".getBytes()));
     }
 
     @Test
     public void testSchemaContentIsEmptyString() {
-        converter.configure(Collections.singletonMap(JsonConverterConfig.SCHEMA_CONTENT_CONFIG, ""), false);
+        converter.configure(Map.of(JsonConverterConfig.SCHEMA_CONTENT_CONFIG, ""), false);
         assertEquals(new SchemaAndValue(Schema.STRING_SCHEMA, "foo-bar-baz"), converter.toConnectData(TOPIC, "{ \"schema\": { \"type\": \"string\" }, \"payload\": \"foo-bar-baz\" }".getBytes()));
     }
 
     @Test
     public void testSchemaContentValidSchema() {
-        converter.configure(Collections.singletonMap(JsonConverterConfig.SCHEMA_CONTENT_CONFIG, "{ \"type\": \"string\" }"), false);
+        converter.configure(Map.of(JsonConverterConfig.SCHEMA_CONTENT_CONFIG, "{ \"type\": \"string\" }"), false);
         assertEquals(new SchemaAndValue(Schema.STRING_SCHEMA, "foo-bar-baz"), converter.toConnectData(TOPIC, "\"foo-bar-baz\"".getBytes()));
     }
 
     @Test
     public void testSchemaContentInValidSchema() {
         assertThrows(
             DataException.class,
-            () -> converter.configure(Collections.singletonMap(JsonConverterConfig.SCHEMA_CONTENT_CONFIG, "{ \"string\" }"), false),
+            () -> converter.configure(Map.of(JsonConverterConfig.SCHEMA_CONTENT_CONFIG, "{ \"string\" }"), false),
             " Provided schema is invalid , please recheck the schema you have provided");
     }
 
     @Test
     public void testSchemaContentLooksLikeSchema() {
-        converter.configure(Collections.singletonMap(JsonConverterConfig.SCHEMA_CONTENT_CONFIG, "{ \"type\": \"struct\", \"fields\": [{\"field\": \"schema\", \"type\": \"struct\",\"fields\": [{\"field\": \"type\", \"type\": \"string\" }]}, {\"field\": \"payload\", \"type\": \"string\"}]}"), false);
+        converter.configure(Map.of(JsonConverterConfig.SCHEMA_CONTENT_CONFIG, "{ \"type\": \"struct\", \"fields\": [{\"field\": \"schema\", \"type\": \"struct\",\"fields\": [{\"field\": \"type\", \"type\": \"string\" }]}, {\"field\": \"payload\", \"type\": \"string\"}]}"), false);
         SchemaAndValue connectData = converter.toConnectData(TOPIC, "{ \"schema\": { \"type\": \"string\" }, \"payload\": \"foo-bar-baz\" }".getBytes());
         assertEquals("foo-bar-baz", ((Struct) connectData.value()).getString("payload"));
     }
 
+    @ParameterizedTest
+    @ValueSource(strings = {
+        "{ }",
+        "{ \"wrong\": \"schema\" }",
+        "{ \"schema\": { \"type\": \"string\" } }",
+        "{ \"payload\": \"foo-bar-baz\" }",
+        "{ \"schema\": { \"type\": \"string\" }, \"payload\": \"foo-bar-baz\", \"extra\": \"field\" }",
+    })
+    public void testNullSchemaContentWithWrongConnectDataValue(String value) {
+        converter.configure(Map.of(), false);
+        assertThrows(
+                DataException.class,
+                () -> converter.toConnectData(TOPIC, value.getBytes()));
+    }
+
     private JsonNode parse(byte[] json) {
         try {
             return objectMapper.readTree(json);