[MINOR] Handle cases of malformed records when converting to json (apache#10943)

Author: the-other-tim-brown

hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/HoodieSparkUtils.scala

@@ -223,7 +223,7 @@ object HoodieSparkUtils extends SparkAdapterSupport with SparkVersionsSupport wi
         val transform: GenericRecord => Either[GenericRecord, String] = record => try {
           Left(HoodieAvroUtils.rewriteRecordDeep(record, schema, true))
         } catch {
-          case _: Throwable => Right(HoodieAvroUtils.avroToJsonString(record, false))
+          case _: Throwable => Right(HoodieAvroUtils.safeAvroToJsonString(record))
         }
         recs.map(transform)
       }

hudi-common/src/main/java/org/apache/hudi/avro/HoodieAvroUtils.java

@@ -197,6 +197,20 @@ public static String avroToJsonString(GenericRecord record, boolean pretty) thro
     return avroToJsonHelper(record, pretty).toString();
   }
 
+  /**
+   * Convert a given avro record to a JSON string. If the record contents are invalid, return the record.toString().
+   * Use this method over {@link HoodieAvroUtils#avroToJsonString} when simply trying to print the record contents without any guarantees around their correctness.
+   * @param record The GenericRecord to convert
+   * @return a JSON string
+   */
+  public static String safeAvroToJsonString(GenericRecord record) {
+    try {
+      return avroToJsonString(record, false);
+    } catch (Exception e) {
+      return record.toString();
+    }
+  }
+
   /**
    * Convert a given avro record to json and return the encoded bytes.
    *

hudi-common/src/test/java/org/apache/hudi/avro/TestHoodieAvroUtils.java

@@ -629,4 +629,27 @@ public void testAddMetadataFields() {
     assertEquals("custom_schema_property_value", schemaWithMetadata.getProp("custom_schema_property"));
     assertEquals("value", originalFieldsInUpdatedSchema.get(0).getProp("custom_field_property"));
   }
+
+  @Test
+  void testSafeAvroToJsonStringMissingRequiredField() {
+    Schema schema = new Schema.Parser().parse(EXAMPLE_SCHEMA);
+    GenericRecord record = new GenericData.Record(schema);
+    record.put("non_pii_col", "val1");
+    record.put("pii_col", "val2");
+    record.put("timestamp", 3.5);
+    String jsonString = HoodieAvroUtils.safeAvroToJsonString(record);
+    assertEquals("{\"timestamp\": 3.5, \"_row_key\": null, \"non_pii_col\": \"val1\", \"pii_col\": \"val2\"}", jsonString);
+  }
+
+  @Test
+  void testSafeAvroToJsonStringBadDataType() {
+    Schema schema = new Schema.Parser().parse(EXAMPLE_SCHEMA);
+    GenericRecord record = new GenericData.Record(schema);
+    record.put("non_pii_col", "val1");
+    record.put("_row_key", "key");
+    record.put("pii_col", "val2");
+    record.put("timestamp", "foo");
+    String jsonString = HoodieAvroUtils.safeAvroToJsonString(record);
+    assertEquals("{\"timestamp\": \"foo\", \"_row_key\": \"key\", \"non_pii_col\": \"val1\", \"pii_col\": \"val2\"}", jsonString);
+  }
 }

hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/ErrorEvent.java

@@ -19,6 +19,8 @@
 
 package org.apache.hudi.utilities.streamer;
 
+import java.util.Objects;
+
 /**
  * Error event is an event triggered during write or processing failure of a record.
  */
@@ -40,6 +42,23 @@ public ErrorReason getReason() {
     return reason;
   }
 
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    ErrorEvent<?> that = (ErrorEvent<?>) o;
+    return reason == that.reason && Objects.equals(payload, that.payload);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(reason, payload);
+  }
+
   /**
    * The reason behind write or processing failure of a record
    */

hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieStreamerUtils.java

@@ -167,7 +167,7 @@ private static Either<HoodieRecord, String> generateErrorRecordOrThrowException(
       }
     }
     try {
-      return Either.right(HoodieAvroUtils.avroToJsonString(genRec, false));
+      return Either.right(HoodieAvroUtils.safeAvroToJsonString(genRec));
     } catch (Exception ex) {
       throw new HoodieException("Failed to convert illegal record to json", ex);
     }

hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/TestHoodieStreamerUtils.java

@@ -29,17 +29,18 @@
 import org.apache.avro.Schema;
 import org.apache.avro.generic.GenericData;
 import org.apache.avro.generic.GenericRecord;
-import org.apache.spark.SparkException;
 import org.apache.spark.api.java.JavaRDD;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.EnumSource;
+import org.mockito.ArgumentCaptor;
 import org.mockito.Mockito;
 
 import java.util.Collections;
+import java.util.List;
 
-import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.doNothing;
 
 /**
  * Tests {@link HoodieStreamerUtils}.
@@ -73,12 +74,13 @@ public void testCreateHoodieRecordsWithError(HoodieRecordType recordType) {
     TypedProperties props = new TypedProperties();
     SchemaProvider schemaProvider = new SimpleSchemaProvider(jsc, schema, props);
     BaseErrorTableWriter errorTableWriter = Mockito.mock(BaseErrorTableWriter.class);
-    SparkException exception = assertThrows(
-        SparkException.class,
-        () -> HoodieStreamerUtils.createHoodieRecords(cfg, props, Option.of(recordRdd),
-                schemaProvider, recordType, false, "000", Option.of(errorTableWriter))
-            .get().collect()
-    );
-    assertTrue(exception.getMessage().contains("Failed to convert illegal record to json"));
+    ArgumentCaptor<JavaRDD<?>> errorEventCaptor = ArgumentCaptor.forClass(JavaRDD.class);
+    doNothing().when(errorTableWriter).addErrorEvents(errorEventCaptor.capture());
+    HoodieStreamerUtils.createHoodieRecords(cfg, props, Option.of(recordRdd),
+                schemaProvider, recordType, false, "000", Option.of(errorTableWriter));
+    List<ErrorEvent<String>> actualErrorEvents = (List<ErrorEvent<String>>) errorEventCaptor.getValue().collect();
+    ErrorEvent<String> expectedErrorEvent = new ErrorEvent<>("{\"timestamp\": 1000, \"_row_key\": \"key1\", \"partition_path\": \"path1\", \"rider\": null, \"driver\": \"driver\"}",
+        ErrorEvent.ErrorReason.RECORD_CREATION);
+    assertEquals(Collections.singletonList(expectedErrorEvent), actualErrorEvents);
   }
 }