Defect/add tombstone record testcase

src/test/java/com/microsoft/azure/kusto/kafka/connect/sink/E2ETest.java

@@ -87,6 +87,23 @@ public void testE2EJson() throws URISyntaxException, DataClientException, DataSe
         }
     }
 
+    @Test
+    public void testE2EJsonTombstone() throws URISyntaxException, DataClientException, DataServiceException {
+        String dataFormat = "json";
+        IngestionMapping.IngestionMappingKind ingestionMappingKind = IngestionMapping.IngestionMappingKind.JSON;
+        String mapping = "{\"column\":\"ColA\", \"DataType\":\"string\", \"Properties\":{\"Path\":\"$.ColA\"}}," +
+                "{\"column\":\"ColB\", \"DataType\":\"int\", \"Properties\":{\"Path\":\"$.ColB\"}},";
+        String[] messages = new String[] {"{'ColA': 'first field a', 'ColB': '11'}","{'ColA': 'first field a', 'ColB': '11'}", null};
+        List<byte[]> messagesBytes = new ArrayList<>();
+        messagesBytes.add(messages[0].getBytes());
+        messagesBytes.add(null);
+        long flushInterval = 100;
+
+        if (!executeTest(dataFormat, ingestionMappingKind, mapping, messagesBytes, flushInterval, false)) {
+            Assertions.fail("Test failed");
+        }
+    }
+
     @Test
     public void testE2EAvro() throws URISyntaxException, DataClientException, DataServiceException {
         String dataFormat = "avro";

src/test/java/com/microsoft/azure/kusto/kafka/connect/sink/formatWriter/JsonRecordWriterProviderTest.java

@@ -11,6 +11,7 @@
 import java.util.Map;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
 
 // TODO: Significant duplication among these 4 classes
 public class JsonRecordWriterProviderTest {
@@ -39,4 +40,33 @@ public void testJsonData() throws IOException {
         }
         file.delete();
     }
+
+    @Test
+    public void testJsonDataNull() throws IOException {
+        List<SinkRecord> records = new ArrayList<>();
+        for (int i = 0; i < 10; i++) {
+            Map<String, Integer> map = new HashMap<>();
+            map.put("hello", i);
+            records.add(new SinkRecord("mytopic", 0, null, null, null, map, i));
+        }
+        records.add(new SinkRecord("mytopic", 0, null, null, null, null, 11));
+        File file = new File("abc.json");
+        JsonRecordWriterProvider jsonWriter = new JsonRecordWriterProvider();
+        OutputStream out = new FileOutputStream(file);
+        RecordWriter rd = jsonWriter.getRecordWriter(file.getPath(), out);
+        for (SinkRecord record : records) {
+            rd.write(record);
+        }
+        rd.commit();
+        BufferedReader br = new BufferedReader(new FileReader(file));
+        for(int i = 0; i <= 10; i++){
+            String st = br.readLine();
+            if(i == 10){
+                assertEquals(st,"null");
+            }else {
+                assertEquals(st, String.format("{\"hello\":%s}", i));
+            }
+        }
+        file.delete();
+    }
 }