BufferedRecordMergerFactory throwing Utf8 cannot be cast to Java string error

[voonhous]

Bug Description

What happened:
Running the integration test for org.apache.hudi.integ.ITTestHoodieDemo#testParquetDemo throws the error below:

What you expected:

Steps to reproduce:
1.
2.
3.

Environment

Hudi version: 1.2.0-SNAPSHOT
Query engine: (Spark/Flink/Trino etc) Spark
Relevant configs:

/bin/bash -c spark-submit --class org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer /var/hoodie/ws/docker/hoodie/hadoop/hive_base/target/hoodie-utilities.jar \ 
    --table-type COPY_ON_WRITE \
    --base-file-format PARQUET \
    --source-class org.apache.hudi.utilities.sources.JsonDFSSource \
    --source-ordering-field ts \
    --target-base-path /user/hive/warehouse/stock_ticks_cow \
    --target-table stock_ticks_cow \
    --props /var/demo/config/dfs-source.properties \
    --schemaprovider-class org.apache.hudi.utilities.schema.FilebasedSchemaProvider \
    --enable-hive-sync \
    --hoodie-conf hoodie.datasource.hive_sync.jdbcurl=jdbc:hive2://hiveserver:10000/ 
    --hoodie-conf hoodie.datasource.hive_sync.partition_extractor_class=org.apache.hudi.hive.SlashEncodedDayPartitionValueExtractor  \
    --hoodie-conf hoodie.datasource.hive_sync.username=hive  \
    --hoodie-conf hoodie.datasource.hive_sync.password=hive  \
    --hoodie-conf hoodie.datasource.hive_sync.partition_fields=dt  \
    --hoodie-conf hoodie.datasource.hive_sync.database=default  \
    --hoodie-conf hoodie.datasource.hive_sync.table=stock_ticks_cow

Logs and Stack Trace

java.lang.ClassCastException: class org.apache.avro.util.Utf8 cannot be cast to class java.lang.String (org.apache.avro.util.Utf8 is in unnamed module of loader 'app'; java.lang.String is in module java.base of loader 'bootstrap')
	at java.lang.String.compareTo(String.java:125) ~[?:?]
	at org.apache.hudi.common.table.read.BufferedRecordMergerFactory.shouldKeepNewerRecord(BufferedRecordMergerFactory.java:492) ~[hoodie-utilities.jar:1.2.0-SNAPSHOT]
	at org.apache.hudi.common.table.read.BufferedRecordMergerFactory.access$100(BufferedRecordMergerFactory.java:41) ~[hoodie-utilities.jar:1.2.0-SNAPSHOT]
	at org.apache.hudi.common.table.read.BufferedRecordMergerFactory$EventTimeRecordMerger.finalMerge(BufferedRecordMergerFactory.java:193) ~[hoodie-utilities.jar:1.2.0-SNAPSHOT]
	at org.apache.hudi.common.table.read.buffer.FileGroupRecordBuffer.hasNextBaseRecord(FileGroupRecordBuffer.java:241) ~[hoodie-utilities.jar:1.2.0-SNAPSHOT]
	at org.apache.hudi.common.table.read.buffer.KeyBasedFileGroupRecordBuffer.hasNextBaseRecord(KeyBasedFileGroupRecordBuffer.java:138) ~[hoodie-utilities.jar:1.2.0-SNAPSHOT]
	at org.apache.hudi.common.table.read.buffer.KeyBasedFileGroupRecordBuffer.doHasNext(KeyBasedFileGroupRecordBuffer.java:147) ~[hoodie-utilities.jar:1.2.0-SNAPSHOT]
	at org.apache.hudi.common.table.read.buffer.FileGroupRecordBuffer.hasNext(FileGroupRecordBuffer.java:152) ~[hoodie-utilities.jar:1.2.0-SNAPSHOT]
	at org.apache.hudi.common.table.read.HoodieFileGroupReader.hasNext(HoodieFileGroupReader.java:247) ~[hoodie-utilities.jar:1.2.0-SNAPSHOT]
	at org.apache.hudi.common.table.read.HoodieFileGroupReader$HoodieFileGroupReaderIterator.hasNext(HoodieFileGroupReader.java:334) ~[hoodie-utilities.jar:1.2.0-SNAPSHOT]
	at org.apache.hudi.common.util.collection.MappingIterator.hasNext(MappingIterator.java:39) ~[hoodie-utilities.jar:1.2.0-SNAPSHOT]
	at org.apache.hudi.io.FileGroupReaderBasedMergeHandle.doMerge(FileGroupReaderBasedMergeHandle.java:270) ~[hoodie-utilities.jar:1.2.0-SNAPSHOT]
	at org.apache.hudi.io.IOUtils.runMerge(IOUtils.java:120) ~[hoodie-utilities.jar:1.2.0-SNAPSHOT]
	at org.apache.hudi.table.action.commit.BaseSparkCommitActionExecutor.handleUpdate(BaseSparkCommitActionExecutor.java:392) ~[hoodie-utilities.jar:1.2.0-SNAPSHOT]
	at org.apache.hudi.table.action.commit.BaseSparkCommitActionExecutor.handleUpsertPartition(BaseSparkCommitActionExecutor.java:358) ~[hoodie-utilities.jar:1.2.0-SNAPSHOT]
	at org.apache.hudi.table.action.commit.BaseSparkCommitActionExecutor.lambda$mapPartitionsAsRDD$e664f7e$1(BaseSparkCommitActionExecutor.java:298) ~[hoodie-utilities.jar:1.2.0-SNAPSHOT]
	at org.apache.spark.api.java.JavaRDDLike.$anonfun$mapPartitionsWithIndex$1(JavaRDDLike.scala:102) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.api.java.JavaRDDLike.$anonfun$mapPartitionsWithIndex$1$adapted(JavaRDDLike.scala:102) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.rdd.RDD.$anonfun$mapPartitionsWithIndex$2(RDD.scala:910) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.rdd.RDD.$anonfun$mapPartitionsWithIndex$2$adapted(RDD.scala:910) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:367) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.rdd.RDD.iterator(RDD.scala:331) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:367) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.rdd.RDD.$anonfun$getOrCompute$1(RDD.scala:381) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.storage.BlockManager.$anonfun$getOrElseUpdate$1(BlockManager.scala:1372) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.storage.BlockManager.$anonfun$doPutIterator$1(BlockManager.scala:1614) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.storage.BlockManager.org$apache$spark$storage$BlockManager$$doPut(BlockManager.scala:1524) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.storage.BlockManager.doPutIterator(BlockManager.scala:1588) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.storage.BlockManager.getOrElseUpdate(BlockManager.scala:1389) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.storage.BlockManager.getOrElseUpdateRDDBlock(BlockManager.scala:1343) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.rdd.RDD.getOrCompute(RDD.scala:379) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.rdd.RDD.iterator(RDD.scala:329) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:367) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.rdd.RDD.iterator(RDD.scala:331) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:104) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:54) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.TaskContext.runTaskWithListeners(TaskContext.scala:166) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.scheduler.Task.run(Task.scala:141) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$4(Executor.scala:620) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally(SparkErrorUtils.scala:64) [spark-common-utils_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally$(SparkErrorUtils.scala:61) [spark-common-utils_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:94) [spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:623) [spark-core_2.12-3.5.3.jar:3.5.3]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) [?:?]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) [?:?]
	at java.lang.Thread.run(Thread.java:829) [?:?]
26/01/18 13:39:49 WARN BlockManager: Putting block rdd_36_0 failed due to exception org.apache.hudi.exception.HoodieUpsertException: Error upserting bucketType UPDATE for partition :0.
26/01/18 13:39:49 WARN BlockManager: Block rdd_36_0 could not be removed as it was not found on disk or in memory
26/01/18 13:39:49 ERROR Executor: Exception in task 0.0 in stage 14.0 (TID 17)
org.apache.hudi.exception.HoodieUpsertException: Error upserting bucketType UPDATE for partition :0
	at org.apache.hudi.table.action.commit.BaseSparkCommitActionExecutor.handleUpsertPartition(BaseSparkCommitActionExecutor.java:365) ~[hoodie-utilities.jar:1.2.0-SNAPSHOT]
	at org.apache.hudi.table.action.commit.BaseSparkCommitActionExecutor.lambda$mapPartitionsAsRDD$e664f7e$1(BaseSparkCommitActionExecutor.java:298) ~[hoodie-utilities.jar:1.2.0-SNAPSHOT]
	at org.apache.spark.api.java.JavaRDDLike.$anonfun$mapPartitionsWithIndex$1(JavaRDDLike.scala:102) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.api.java.JavaRDDLike.$anonfun$mapPartitionsWithIndex$1$adapted(JavaRDDLike.scala:102) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.rdd.RDD.$anonfun$mapPartitionsWithIndex$2(RDD.scala:910) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.rdd.RDD.$anonfun$mapPartitionsWithIndex$2$adapted(RDD.scala:910) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:367) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.rdd.RDD.iterator(RDD.scala:331) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:367) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.rdd.RDD.$anonfun$getOrCompute$1(RDD.scala:381) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.storage.BlockManager.$anonfun$getOrElseUpdate$1(BlockManager.scala:1372) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.storage.BlockManager.$anonfun$doPutIterator$1(BlockManager.scala:1614) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.storage.BlockManager.org$apache$spark$storage$BlockManager$$doPut(BlockManager.scala:1524) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.storage.BlockManager.doPutIterator(BlockManager.scala:1588) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.storage.BlockManager.getOrElseUpdate(BlockManager.scala:1389) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.storage.BlockManager.getOrElseUpdateRDDBlock(BlockManager.scala:1343) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.rdd.RDD.getOrCompute(RDD.scala:379) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.rdd.RDD.iterator(RDD.scala:329) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:367) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.rdd.RDD.iterator(RDD.scala:331) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:104) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:54) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.TaskContext.runTaskWithListeners(TaskContext.scala:166) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.scheduler.Task.run(Task.scala:141) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$4(Executor.scala:620) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally(SparkErrorUtils.scala:64) ~[spark-common-utils_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally$(SparkErrorUtils.scala:61) ~[spark-common-utils_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:94) ~[spark-core_2.12-3.5.3.jar:3.5.3]
	at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:623) [spark-core_2.12-3.5.3.jar:3.5.3]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) [?:?]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) [?:?]
	at java.lang.Thread.run(Thread.java:829) [?:?]
Caused by: java.lang.ClassCastException: class org.apache.avro.util.Utf8 cannot be cast to class java.lang.String (org.apache.avro.util.Utf8 is in unnamed module of loader 'app'; java.lang.String is in module java.base of loader 'bootstrap')
	at java.lang.String.compareTo(String.java:125) ~[?:?]
	at org.apache.hudi.common.table.read.BufferedRecordMergerFactory.shouldKeepNewerRecord(BufferedRecordMergerFactory.java:492) ~[hoodie-utilities.jar:1.2.0-SNAPSHOT]
	at org.apache.hudi.common.table.read.BufferedRecordMergerFactory.access$100(BufferedRecordMergerFactory.java:41) ~[hoodie-utilities.jar:1.2.0-SNAPSHOT]
	at org.apache.hudi.common.table.read.BufferedRecordMergerFactory$EventTimeRecordMerger.finalMerge(BufferedRecordMergerFactory.java:193) ~[hoodie-utilities.jar:1.2.0-SNAPSHOT]
	at org.apache.hudi.common.table.read.buffer.FileGroupRecordBuffer.hasNextBaseRecord(FileGroupRecordBuffer.java:241) ~[hoodie-utilities.jar:1.2.0-SNAPSHOT]
	at org.apache.hudi.common.table.read.buffer.KeyBasedFileGroupRecordBuffer.hasNextBaseRecord(KeyBasedFileGroupRecordBuffer.java:138) ~[hoodie-utilities.jar:1.2.0-SNAPSHOT]
	at org.apache.hudi.common.table.read.buffer.KeyBasedFileGroupRecordBuffer.doHasNext(KeyBasedFileGroupRecordBuffer.java:147) ~[hoodie-utilities.jar:1.2.0-SNAPSHOT]
	at org.apache.hudi.common.table.read.buffer.FileGroupRecordBuffer.hasNext(FileGroupRecordBuffer.java:152) ~[hoodie-utilities.jar:1.2.0-SNAPSHOT]
	at org.apache.hudi.common.table.read.HoodieFileGroupReader.hasNext(HoodieFileGroupReader.java:247) ~[hoodie-utilities.jar:1.2.0-SNAPSHOT]
	at org.apache.hudi.common.table.read.HoodieFileGroupReader$HoodieFileGroupReaderIterator.hasNext(HoodieFileGroupReader.java:334) ~[hoodie-utilities.jar:1.2.0-SNAPSHOT]
	at org.apache.hudi.common.util.collection.MappingIterator.hasNext(MappingIterator.java:39) ~[hoodie-utilities.jar:1.2.0-SNAPSHOT]
	at org.apache.hudi.io.FileGroupReaderBasedMergeHandle.doMerge(FileGroupReaderBasedMergeHandle.java:270) ~[hoodie-utilities.jar:1.2.0-SNAPSHOT]
	at org.apache.hudi.io.IOUtils.runMerge(IOUtils.java:120) ~[hoodie-utilities.jar:1.2.0-SNAPSHOT]
	at org.apache.hudi.table.action.commit.BaseSparkCommitActionExecutor.handleUpdate(BaseSparkCommitActionExecutor.java:392) ~[hoodie-utilities.jar:1.2.0-SNAPSHOT]
	at org.apache.hudi.table.action.commit.BaseSparkCommitActionExecutor.handleUpsertPartition(BaseSparkCommitActionExecutor.java:358) ~[hoodie-utilities.jar:1.2.0-SNAPSHOT]
	... 35 more

[voonhous]

Leaving this here for reference in the future. The issue is from incoming records having an orderingValue that is of java.lang.String type where it should be Avro UTF8 type.

[danny0405]

The issue is from incoming records having an orderingValue that is of java.lang.String type where it should be Avro UTF8 type.

We now unify the ordering value comparison to use the engine native type, for Spark, it should be the spark UTF8String instead of avro.

[voonhous]

I understand that we unify ordering value comparison to use engine native types for spark.

However, what i'm seeing here is that for the same workflow, one is presented as UTF8, while the other is java.lang.String. So i'm wondering if there's somewhere we missed out.

[voonhous]

Summary of root cause

Found the root cause:

In StreamSync we do these things (which is used by DeltaStreamer, which IIRC is renamed to something else in current repo)

NOTE: Let us only consider the case where HoodieRecordType is HoodieRecordType.AVRO.

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

Lines 508 to 533 in a55bc00

	public Pair<Option<String>, JavaRDD<WriteStatus>> syncOnce() throws IOException {
	Pair<Option<String>, JavaRDD<WriteStatus>> result = null;
	Timer.Context overallTimerContext = metrics.getOverallTimerContext();
	try {
	// Refresh Timeline
	HoodieTableMetaClient metaClient = initializeMetaClientAndRefreshTimeline();
	Pair<InputBatch, Boolean> inputBatchAndUseRowWriter = readFromSource(metaClient);
	if (inputBatchAndUseRowWriter != null) {
	InputBatch inputBatch = inputBatchAndUseRowWriter.getLeft();
	boolean useRowWriter = inputBatchAndUseRowWriter.getRight();
	initializeWriteClientAndRetryTableServices(inputBatch, metaClient);
	result = writeToSinkAndDoMetaSync(metaClient, inputBatch, useRowWriter, metrics, overallTimerContext);
	}
	// refresh schemas if need be before next batch
	if (schemaProvider != null) {
	schemaProvider.refresh();
	}
	metrics.updateStreamerSyncMetrics(System.currentTimeMillis());
	return result;
	} finally {
	this.formatAdapter.getSource().releaseResources();
	}
	}

1. readFromSource
   • Invokes fetchNewDataInAvroFormat
   • Invokes AvroConvertor.fromJson()
   • Uses MercifulJsonConverter which converts Strings to java.lang.String
2. writeToSinkAndDoMetaSync
   • Does required merging using BufferedRecordMergerFactory
   • Commits data
   • Performs metasync after commit is successful

MercifulJsonConverter code that returns java.lang.String:

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

Lines 441 to 456 in a55bc00

	private static class StringProcessor extends JsonFieldProcessor {
	private static final ObjectMapper STRING_MAPPER = new ObjectMapper();
	@Override
	public Pair<Boolean, Object> convert(Object value, String name, HoodieSchema schema) {
	if (value instanceof String) {
	return Pair.of(true, value);
	} else {
	try {
	return Pair.of(true, STRING_MAPPER.writeValueAsString(value));
	} catch (IOException ex) {
	return Pair.of(false, null);
	}
	}
	}
	}

Line 447 returns the value as String.
Hence the incoming records that is invoked using readFromSource, the String columns (orderingVal column) will hence be in java.lang.String.

But for records with String columns that are read from baseFile, they will be read out as Avro's Utf8, hence, when performing the merging, it triggers the merge error.

TLDR of the root cause: MercifulJsonConverter.StringProcessor returns java.lang.String directly instead of wrapping it in Utf8. This is inconsistent with Avro's standard behavior where strings should be Utf8 by default.

Dataflow references

1. JSON sources -> SourceFormatAdapter.fetchNewDataInAvroFormat() -> AvroConvertor.fromJson() -> MercifulJsonConverter.convert() -> java.lang.String (via StringProcessor which does NOT convert to Utf8)
2. ROW sources -> HoodieSparkUtils.createRdd() -> AvroSerializer -> Utf8 (Spark's AvroSerializer: new Utf8(getter.getUTF8String(ordinal).getBytes))
3. AVRO sources → Returns directly what the source provides (typically Utf8 when reading standard Avro files)

RecordMerger used:
HoodieAvroRecordMerger

[voonhous]

Okay, found the configs that will trigger this:

1. COW table
2. No data transformers
3. HoodieDeltaStreamer entrypoint
4. RecordType = HoodieRecordType.Avro -> HoodieAvroMerger used
5. sourceOrderingField/preCombineField is on a String column

[voonhous]

Use this code snippet to reproduce the error, add this into TestBufferedRecords.

  /**
   * This test verifies the bug where records coming from JSON source (via MercifulJsonConverter)
   * have string fields stored as java.lang.String, while records from other paths (e.g., Spark DataFrame)
   * have string fields stored as Utf8 is fixed. When these records are compared during merging
   * (e.g., in EVENT_TIME_ORDERING mode), a ClassCastException occurs because
   * String.compareTo() expects a String argument, not Utf8.
   *
   * <p>Data flow paths:
   * <ul>
   *   <li>JSON source (HoodieStreamer): MercifulJsonConverter -> GenericRecord with java.lang.String
   *       -> HoodieAvroUtils.getNestedFieldVal extracts java.lang.String as ordering value</li>
   *   <li>Spark DataFrame (HoodieCreateRecordUtils): AvroSerializer -> GenericRecord with Utf8
   *       -> HoodieAvroUtils.getNestedFieldVal extracts Utf8 as ordering value</li>
   * </ul>
   *
   * <p>The root cause is in MercifulJsonConverter.StringProcessor which returns raw java.lang.String
   * instead of wrapping it in Utf8.
   *
   * @see org.apache.hudi.avro.MercifulJsonConverter
   * @see org.apache.hudi.utilities.streamer.HoodieStreamerUtils#createHoodieRecords
   */
  @Test
  void testStringVsUtf8OrderingValueClassCastException() {
    // Schema with a string field that will be used as ordering field
    String schemaString = "{\"type\": \"record\", \"name\": \"rec\", \"fields\": ["
        + "{\"name\": \"_row_key\", \"type\": \"string\"},"
        + "{\"name\": \"partition_path\", \"type\": \"string\"},"
        + "{\"name\": \"ordering_field\", \"type\": \"string\"}"
        + "]}";
    HoodieSchema schema = HoodieSchema.parse(schemaString);

    // =========================================================================
    // Simulate JSON source path (HoodieStreamer with JSON source):
    // MercifulJsonConverter.convert() -> GenericRecord with java.lang.String
    // =========================================================================
    String jsonRecord = "{\"_row_key\": \"key1\", \"partition_path\": \"path1\", \"ordering_field\": \"2024-01-01\"}";
    MercifulJsonConverter jsonConverter = new MercifulJsonConverter();
    GenericRecord jsonSourceRecord = jsonConverter.convert(jsonRecord, schema);

    // Extract ordering value the same way HoodieStreamerUtils does:
    // OrderingValues.create(orderingFieldsStr.split(","),
    //    field -> (Comparable) HoodieAvroUtils.getNestedFieldVal(gr, field, false, useConsistentLogicalTimestamp))
    Object jsonOrderingValue = HoodieAvroUtils.getNestedFieldVal(jsonSourceRecord, "ordering_field", false, false);

    // Verify: JSON source does not produce java.lang.String (THIS IS THE BUG)
    assertInstanceOf(Utf8.class, jsonOrderingValue,
        "JSON source (MercifulJsonConverter) should produce org.apache.avro.util.Utf8 for string fields");
    assertEquals(new Utf8("2024-01-01"), jsonOrderingValue);

    // =========================================================================
    // Simulate Spark DataFrame path (HoodieCreateRecordUtils):
    // AvroSerializer produces GenericRecord with Utf8 for string fields
    // =========================================================================
    GenericRecord sparkSourceRecord = new GenericData.Record(schema.toAvroSchema());
    sparkSourceRecord.put("_row_key", new Utf8("key2"));
    sparkSourceRecord.put("partition_path", new Utf8("path2"));
    sparkSourceRecord.put("ordering_field", new Utf8("2024-01-02"));

    Object sparkOrderingValue = HoodieAvroUtils.getNestedFieldVal(sparkSourceRecord, "ordering_field", false, false);

    // Verify: Spark source produces Utf8 (correct behavior)
    assertInstanceOf(Utf8.class, sparkOrderingValue,
        "Spark DataFrame source should produce Utf8 for string fields");
    assertEquals(new Utf8("2024-01-02"), sparkOrderingValue);

    // =========================================================================
    // Create BufferedRecords simulating what happens during EVENT_TIME_ORDERING merge:
    // - oldRecord: previously written via Spark DataFrame path (Utf8 ordering value)
    // - newRecord: incoming from JSON source via HoodieStreamer (java.lang.String ordering value)
    // =========================================================================
    BufferedRecord<GenericRecord> oldRecord = new BufferedRecord<>(
        "key2", (Comparable) sparkOrderingValue, sparkSourceRecord, 1, null);
    BufferedRecord<GenericRecord> newRecord = new BufferedRecord<>(
        "key1", (Comparable) jsonOrderingValue, jsonSourceRecord, 1, null);

    // Get ordering values as they would be accessed in BufferedRecordMergerFactory.shouldKeepNewerRecord()
    Comparable newOrderingVal = newRecord.getOrderingValue();
    Comparable oldOrderingVal = oldRecord.getOrderingValue();

    // =========================================================================
    // Demonstrate that ClassCastException bug is fixed
    // This is what happens in BufferedRecordMergerFactory.shouldKeepNewerRecord():
    //   return newRecord.getOrderingValue().compareTo(oldRecord.getOrderingValue()) >= 0;
    //
    // When newOrderingVal should not be in java.lang.String and oldOrderingVal is in Utf8:
    //   String("2024-01-01").compareTo(Utf8("2024-01-02")) throws ClassCastException
    // =========================================================================
    assertDoesNotThrow(() -> newOrderingVal.compareTo(oldOrderingVal),
        "newOrderingVal and oldOrderingVal should be of the same type");
  }

[voonhous]

If we make MercifulJsonConverter return a UTF8, this error will be thrown:

[ERROR]   Run 4: TestJsonKafkaSource.testJsonKafkaSourceWithEncodedDecimals » Spark Job aborted due to stage failure: Task 0 in stage 2.0 failed 1 times, most recent failure: Lost task 0.0 in stage 2.0 (TID 6) (be008c538ca3 executor driver): org.apache.spark.SparkRuntimeException: Error while encoding: java.lang.RuntimeException: org.apache.avro.util.Utf8 is not a valid external type for schema of string
assertnotnull(validateexternaltype(getexternalrowfield(assertnotnull(input[0, org.apache.spark.sql.Row, true]), 0, timestamp), LongType, ObjectType(class java.lang.Long)).longValue) AS timestamp#38958L
assertnotnull(staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, validateexternaltype(getexternalrowfield(assertnotnull(input[0, org.apache.spark.sql.Row, true]), 1, _row_key), StringType, ObjectType(class java.lang.String)), true, false, true)) AS _row_key#38959
assertnotnull(staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, validateexternaltype(getexternalrowfield(assertnotnull(input[0, org.apache.spark.sql.Row, true]), 2, rider), StringType, ObjectType(class java.lang.String)), true, false, true)) AS rider#38960
assertnotnull(CheckOverflow(staticinvoke(class org.apache.spark.sql.types.Decimal$, DecimalType(10,6), fromDecimal, validateexternaltype(getexternalrowfield(assertnotnull(input[0, org.apache.spark.sql.Row, true]), 3, decfield), DecimalType(10,6), ObjectType(class java.lang.Object)), true, false, true), DecimalType(10,6))) AS decfield#38961
assertnotnull(CheckOverflow(staticinvoke(class org.apache.spark.sql.types.Decimal$, DecimalType(4,2), fromDecimal, validateexternaltype(getexternalrowfield(assertnotnull(input[0, org.apache.spark.sql.Row, true]), 4, lowprecision), DecimalType(4,2), ObjectType(class java.lang.Object)), true, false, true), DecimalType(4,2))) AS lowprecision#38962
assertnotnull(CheckOverflow(staticinvoke(class org.apache.spark.sql.types.Decimal$, DecimalType(32,12), fromDecimal, validateexternaltype(getexternalrowfield(assertnotnull(input[0, org.apache.spark.sql.Row, true]), 5, highprecision), DecimalType(32,12), ObjectType(class java.lang.Object)), true, false, true), DecimalType(32,12))) AS highprecision#38963
assertnotnull(staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, validateexternaltype(getexternalrowfield(assertnotnull(input[0, org.apache.spark.sql.Row, true]), 6, driver), StringType, ObjectType(class java.lang.String)), true, false, true)) AS driver#38964
assertnotnull(validateexternaltype(getexternalrowfield(assertnotnull(input[0, org.apache.spark.sql.Row, true]), 7, fare), DoubleType, ObjectType(class java.lang.Double)).doubleValue) AS fare#38965
assertnotnull(validateexternaltype(getexternalrowfield(assertnotnull(input[0, org.apache.spark.sql.Row, true]), 8, _hoodie_is_deleted), BooleanType, ObjectType(class java.lang.Boolean)).booleanValue) AS _hoodie_is_deleted#38966.
	at org.apache.spark.sql.errors.QueryExecutionErrors$.expressionEncodingError(QueryExecutionErrors.scala:1416)
	at org.apache.spark.sql.catalyst.encoders.ExpressionEncoder$Serializer.apply(ExpressionEncoder.scala:217)
	at org.apache.spark.sql.catalyst.encoders.ExpressionEncoder$Serializer.apply(ExpressionEncoder.scala:200)
	at scala.collection.Iterator$$anon$10.next(Iterator.scala:461)
	at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.hashAgg_doAggregateWithoutKey_0$(Unknown Source)
	at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)
	at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
	at org.apache.spark.sql.execution.WholeStageCodegenEvaluatorFactory$WholeStageCodegenPartitionEvaluator$$anon$1.hasNext(WholeStageCodegenEvaluatorFactory.scala:43)
	at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460)
	at org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.write(BypassMergeSortShuffleWriter.java:140)
	at org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59)
	at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:104)
	at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:54)
	at org.apache.spark.TaskContext.runTaskWithListeners(TaskContext.scala:166)
	at org.apache.spark.scheduler.Task.run(Task.scala:141)
	at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$4(Executor.scala:620)
	at org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally(SparkErrorUtils.scala:64)
	at org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally$(SparkErrorUtils.scala:61)
	at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:94)
	at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:623)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
	at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: java.lang.RuntimeException: org.apache.avro.util.Utf8 is not a valid external type for schema of string
	at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificUnsafeProjection.writeFields_0_0$(Unknown Source)
	at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificUnsafeProjection.apply(Unknown Source)
	at org.apache.spark.sql.catalyst.encoders.ExpressionEncoder$Serializer.apply(ExpressionEncoder.scala:214)
	... 21 more

Bunch of other errors will be thrown too:

[ERROR] org.apache.hudi.utilities.sources.helpers.TestMercifulJsonToRowConverterJava8Api.basicConversion
[ERROR]   Run 1: TestMercifulJsonToRowConverterJava8Api>TestMercifulJsonToRowConverterBase.basicConversion:100->TestMercifulJsonToRowConverterBase.validateSchemaCompatibility:690 » IllegalArgument The value (John Smith) of the type (org.apache.avro.util.Utf8) cannot be converted to the string type
[ERROR]   Run 2: TestMercifulJsonToRowConverterJava8Api>TestMercifulJsonToRowConverterBase.basicConversion:100->TestMercifulJsonToRowConverterBase.validateSchemaCompatibility:690 » IllegalArgument The value (John Smith) of the type (org.apache.avro.util.Utf8) cannot be converted to the string type
[ERROR]   Run 3: TestMercifulJsonToRowConverterJava8Api>TestMercifulJsonToRowConverterBase.basicConversion:100->TestMercifulJsonToRowConverterBase.validateSchemaCompatibility:690 » IllegalArgument The value (John Smith) of the type (org.apache.avro.util.Utf8) cannot be converted to the string type
[ERROR]   Run 4: TestMercifulJsonToRowConverterJava8Api>TestMercifulJsonToRowConverterBase.basicConversion:100->TestMercifulJsonToRowConverterBase.validateSchemaCompatibility:690 » IllegalArgument The value (John Smith) of the type (org.apache.avro.util.Utf8) cannot be converted to the string type
[INFO] 
[ERROR] org.apache.hudi.utilities.sources.helpers.TestMercifulJsonToRowConverterJava8Api.conversionWithFieldNameAliases
[ERROR]   Run 1: TestMercifulJsonToRowConverterJava8Api>TestMercifulJsonToRowConverterBase.conversionWithFieldNameAliases:627->TestMercifulJsonToRowConverterBase.validateSchemaCompatibility:690 » IllegalArgument The value (John Smith) of the type (org.apache.avro.util.Utf8) cannot be converted to the string type
[ERROR]   Run 2: TestMercifulJsonToRowConverterJava8Api>TestMercifulJsonToRowConverterBase.conversionWithFieldNameAliases:627->TestMercifulJsonToRowConverterBase.validateSchemaCompatibility:690 » IllegalArgument The value (John Smith) of the type (org.apache.avro.util.Utf8) cannot be converted to the string type
[ERROR]   Run 3: TestMercifulJsonToRowConverterJava8Api>TestMercifulJsonToRowConverterBase.conversionWithFieldNameAliases:627->TestMercifulJsonToRowConverterBase.validateSchemaCompatibility:690 » IllegalArgument The value (John Smith) of the type (org.apache.avro.util.Utf8) cannot be converted to the string type
[ERROR]   Run 4: TestMercifulJsonToRowConverterJava8Api>TestMercifulJsonToRowConverterBase.conversionWithFieldNameAliases:627->TestMercifulJsonToRowConverterBase.validateSchemaCompatibility:690 » IllegalArgument The value (John Smith) of the type (org.apache.avro.util.Utf8) cannot be converted to the string type

[voonhous]

Ignore the above exception in #17938 (comment).

The issue was because MercifulJsonToRowConverter extends MercifulJsonConverter. It inherited the Utf8 conversion causing other paths to throw the error we see.

Overriding it again in MercifulJsonToRowConverter to produce the correct String type worked again.