[GitHub] [druid] jihoonson commented on a change in pull request #10383: Fix ingestion failure of pretty-formatted JSON message

[GitBox <gi...@apache.org>]

jihoonson commented on a change in pull request #10383:
URL: https://github.com/apache/druid/pull/10383#discussion_r520018811



##########
File path: extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/KafkaIndexTaskTest.java
##########
@@ -2726,4 +2726,83 @@ public void close()
         null
     );
   }
+
+  @Test(timeout = 60_000L)
+  public void testMultipleLinesJSONText() throws Exception
+  {
+    reportParseExceptions = false;
+    maxParseExceptions = 1000;
+    maxSavedParseExceptions = 2;
+
+    // Insert data
+    try (final KafkaProducer<byte[], byte[]> kafkaProducer = kafkaServer.newProducer()) {
+      kafkaProducer.initTransactions();
+      kafkaProducer.beginTransaction();
+
+      //multiple objects in one Kafka record will yield 2 rows in druid
+      String wellformed = toJsonString(true, "2049", "d2", "y", "10", "22.0", "2.0") +
+                     toJsonString(true, "2049", "d3", "y", "10", "23.0", "3.0");
+
+      //multiple objects in one Kafka record but some objects are in ill-formed format
+      //the whole ProducerRecord will be discarded
+      String illformed = "{\"timestamp\":2049, \"dim1\": \"d4\", \"dim2\":\"x\", \"dimLong\": 10, \"dimFloat\":\"24.0\", \"met1\":\"2.0\" }" +
+                     "{\"timestamp\":2049, \"dim1\": \"d5\", \"dim2\":\"y\", \"dimLong\": 10, \"dimFloat\":\"24.0\", \"met1\":invalidFormat }" +
+                     "{\"timestamp\":2049, \"dim1\": \"d6\", \"dim2\":\"z\", \"dimLong\": 10, \"dimFloat\":\"24.0\", \"met1\":\"3.0\" }";
+
+      ProducerRecord[] producerRecords = new ProducerRecord[]{
+          // pretty formatted
+          new ProducerRecord<>(topic, 0, null, jb(true, "2049", "d1", "y", "10", "20.0", "1.0")),
+          //well-formed
+          new ProducerRecord<>(topic, 0, null, StringUtils.toUtf8(wellformed)),
+          //ill-formed
+          new ProducerRecord<>(topic, 0, null, StringUtils.toUtf8("illformed")),

Review comment:
       Did you intend `StringUtils.toUtf8(illformed)`? The variable `illformed` is not in use.

##########
File path: indexing-service/src/main/java/org/apache/druid/indexing/overlord/sampler/InputSourceSampler.java
##########
@@ -111,38 +113,49 @@ public SamplerResponse sample(
     try (final CloseableIterator<InputRowListPlusRawValues> iterator = reader.sample();
          final IncrementalIndex<Aggregator> index = buildIncrementalIndex(nonNullSamplerConfig, nonNullDataSchema);
          final Closer closer1 = closer) {
-      SamplerResponseRow[] responseRows = new SamplerResponseRow[nonNullSamplerConfig.getNumRows()];
-      int counter = 0, numRowsIndexed = 0;
-
-      while (counter < responseRows.length && iterator.hasNext()) {
-        Map<String, Object> rawColumns = null;
-        try {
-          final InputRowListPlusRawValues inputRowListPlusRawValues = iterator.next();
-
-          if (inputRowListPlusRawValues.getRawValues() != null) {
-            rawColumns = inputRowListPlusRawValues.getRawValues();
-          }
-
-          if (inputRowListPlusRawValues.getParseException() != null) {
-            throw inputRowListPlusRawValues.getParseException();
+      List<SamplerResponseRow> responseRows = new ArrayList<>(nonNullSamplerConfig.getNumRows());
+      int numRowsIndexed = 0;
+
+      while (responseRows.size() < nonNullSamplerConfig.getNumRows() && iterator.hasNext()) {
+        final InputRowListPlusRawValues inputRowListPlusRawValues = iterator.next();
+
+        final List<Map<String, Object>> rawColumnsList = inputRowListPlusRawValues.getRawValuesList();
+
+        final ParseException parseException = inputRowListPlusRawValues.getParseException();
+        if (parseException != null) {
+          if (rawColumnsList != null) {
+            // add all rows to response
+            responseRows.addAll(rawColumnsList.stream()
+                                              .map(rawColumns -> new SamplerResponseRow(rawColumns, null, true, parseException.getMessage()))
+                                              .collect(Collectors.toList()));

Review comment:
       nit: with this change, when a parseException is thrown while parsing a list of rows in one message, those rows in the message will be added as separate rows in `responseRows` but with the same parseException. This can be confusing since the parseException error message seem irrelevant to the associated `rawColumns` in the same `responseRow`. IMO, the better fix in this case would be storing the whole `rawColumnsList` in one `SamplerResponseRow`. Then, the parseException can indicate that it was thrown while parsing one of the rows in `rawColumnsList`. However, this requires a change on the web console side as well. I'm OK with fixing this in a follow-up PR.

##########
File path: indexing-service/src/test/java/org/apache/druid/indexing/overlord/sampler/InputSourceSamplerTest.java
##########
@@ -1060,6 +1062,107 @@ public void testWithFilter() throws IOException
     );
   }
 
+  @Test
+  public void testIndexParseException() throws IOException

Review comment:
       Thanks for adding this test! Similar to https://github.com/apache/druid/pull/10383/files#diff-ef25ac1cc1f275b47b939b65e1d0c8b8b8512aeada52d06b8541b8f381df03eeR2731, could you please add a unit test for sampling a block of multiple JSON strings? The unit test can be run only when `parserType` is `STR_JSON`. [We usually just return in the unit test when the parameter is not what we want to test with](https://github.com/apache/druid/blob/master/processing/src/test/java/org/apache/druid/query/groupby/GroupByQueryRunnerTest.java#L6812-L6814). It would be nice if the test verifies the followings:
   - The sampler response when there is no parseException with a list of multiple JSON strings.
   - The sampler response when there is a parseException thrown while parsing a list of multiple JSON strings. Maybe you can improve this unit test to do it as well.




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org