Unsynchronized `LocalRegistry.clear()` can NPE mid-commit and truncate the completed commit metadata file, permanently breaking the table

[bhavya-ganatra]

Bug Description

What happened:

Under concurrent multi-table writes from a single Spark driver JVM with hoodie.metrics.on=true, a NullPointerException was thrown from inside LocalRegistry.add() while Hudi was serializing the completed deltacommit metadata to S3. The exception surfaced inside DataFileWriter.close(), after the Avro block header had been flushed to the output stream but before the record payload was written. HoodieStorage.createImmutableFileInPath closed the stream while unwinding, so object storage received a truncated but non-empty object at the completed-instant filename.

The result is a permanently broken table: <instant>_<completion>.deltacommit exists and looks like a valid completed instant, but cannot be deserialized. Every subsequent write, compaction, clean, and Spark read of that table fails, because ActiveTimelineV2.getLastCommitMetadataWithValidSchema() throws on an unreadable completed instant rather than skipping it.

Three separate problems compose here:

1. LocalRegistry is not thread-safe. getCounter() is synchronized and does containsKey -> put -> get, but clear() is not synchronized (a bare counters.clear()). A clear() landing between the containsKey and the get makes get() return null, so getCounter(name).add(value) NPEs. In steady state the counter already exists, so the exposed window is containsKey -> get, i.e. every add() call, not only first insertion.

2. clear() is called on a JVM-global registry on every save(). HoodieSparkSqlWriter.cleanup() calls Metrics.shutdownAllMetrics(), and DefaultSource.createRelation invokes cleanup() on both the success path and inside an exception handler that rethrows (effectively try/finally), so it runs on every save(). Metrics.shutdown() -> registerHoodieCommonMetrics() -> Registry.getAllMetrics(flush=true, ...) -> clear() on every registry in the global REGISTRY_MAP, which includes the JVM-static HoodieWrapperFileSystem.METRICS_REGISTRY_DATA/_META that all tables write into via SizeAwareFSDataOutputStream.write(). shutdown() also removes the instance from METRICS_INSTANCE_PER_BASEPATH, so the next batch's Metrics constructor re-registers and clears again.

   Net effect: at least two global registry wipes per table per micro-batch. With ~30 tables written concurrently from one driver, that is ~60 wipes per batch racing against continuous add() calls from all writer threads.

3. A failure during commit-metadata serialization leaves a partial object at an authoritative filename. CommitMetadataSerDe.getInstantWriter streams Avro directly into the storage output stream, so any exception mid-serialization can produce a file that Hudi treats as a completed instant.

What you expected:

• LocalRegistry.getCounter() never returns null, so a metrics operation cannot fail a write.
• One table's save() does not tear down metrics state belonging to other tables in the same JVM.
• A failure while writing commit metadata leaves either a complete file or no file at all — never a partial object at a completed-instant filename.

Steps to reproduce:

1. Set hoodie.metrics.on=true (any reporter type).
2. From a single Spark driver JVM, write to N Hudi tables concurrently on separate threads (N ~= 30 in our case), each via df.write.format("hudi")...save(path), in a loop (we used Structured Streaming micro-batches, but any repeated concurrent write should work).
3. Let it run. Each save() completion triggers Metrics.shutdownAllMetrics(), clearing the JVM-static FS metrics registry while other threads are still calling Registry.add(...) on it from their write paths.
4. Eventually LocalRegistry.getCounter() returns null and the NPE lands inside a commit-metadata write, leaving a truncated .deltacommit. In our environment this happened roughly 3 times over 2 weeks per table set — it is a narrow race, so reproduction is probabilistic.

Environment

Hudi version: 1.1.0 (hudi-spark3.5-bundle_2.12-1.1.0, hudi-aws-bundle-1.1.0)
Query engine: Spark 3.5.6, Scala 2.12. Table type MERGE_ON_READ, table version 9, timeline layout version 2, non-partitioned. Storage: S3 (S3A).
Relevant configs:

hoodie.metrics.on=true
hoodie.metrics.reporter.type=PROMETHEUS_PUSHGATEWAY
hoodie.metrics.executor.enable=false          # so the driver-side registry is LocalRegistry
hoodie.datasource.write.table.type=MERGE_ON_READ
hoodie.metadata.enable=true
hoodie.write.concurrency.mode=optimistic_concurrency_control
hoodie.clean.automatic=false                  # clean/compaction run as separate jobs

Key environmental condition: many Hudi write clients active concurrently inside one driver JVM. A single-writer driver will not hit this.

Logs and Stack Trace

1. The originating NPE (application package names genericized)

ERROR [pool-3-thread-3464] Error writing to Hudi: Cannot invoke
  "org.apache.hudi.common.metrics.Counter.add(long)" because the return value of
  "org.apache.hudi.common.metrics.LocalRegistry.getCounter(String)" is null
java.lang.NullPointerException: Cannot invoke "org.apache.hudi.common.metrics.Counter.add(long)"
  because the return value of "org.apache.hudi.common.metrics.LocalRegistry.getCounter(String)" is null
	at org.apache.hudi.common.metrics.LocalRegistry.add(LocalRegistry.java:48) ~[hudi-aws-bundle-1.1.0.jar:1.1.0]
	at org.apache.hudi.hadoop.fs.HoodieWrapperFileSystem.executeFuncWithTimeAndByteMetrics(HoodieWrapperFileSystem.java:133) ~[hudi-spark3.5-bundle_2.12-1.1.0.jar:1.1.0]
	at org.apache.hudi.hadoop.fs.SizeAwareFSDataOutputStream.write(SizeAwareFSDataOutputStream.java:59) ~[hudi-spark3.5-bundle_2.12-1.1.0.jar:1.1.0]
	at org.apache.avro.file.DataFileWriter$BufferedFileOutputStream$PositionFilter.write(DataFileWriter.java:485) ~[avro-1.11.4.jar:1.11.4]
	at java.io.BufferedOutputStream.write(BufferedOutputStream.java:123) ~[?:?]
	at org.apache.avro.io.DirectBinaryEncoder.writeFixed(DirectBinaryEncoder.java:124) ~[avro-1.11.4.jar:1.11.4]
	at org.apache.avro.file.DataFileStream$DataBlock.writeBlockTo(DataFileStream.java:407) ~[avro-1.11.4.jar:1.11.4]
	at org.apache.avro.file.DataFileWriter.writeBlock(DataFileWriter.java:417) ~[avro-1.11.4.jar:1.11.4]
	at org.apache.avro.file.DataFileWriter.sync(DataFileWriter.java:437) ~[avro-1.11.4.jar:1.11.4]
	at org.apache.avro.file.DataFileWriter.flush(DataFileWriter.java:446) ~[avro-1.11.4.jar:1.11.4]
	at org.apache.avro.file.DataFileWriter.close(DataFileWriter.java:469) ~[avro-1.11.4.jar:1.11.4]
	at org.apache.hudi.common.table.timeline.CommitMetadataSerDe.lambda$getInstantWriter$0(CommitMetadataSerDe.java:52) ~[hudi-spark3.5-bundle_2.12-1.1.0.jar:1.1.0]
	at org.apache.hudi.storage.HoodieStorage.createImmutableFileInPath(HoodieStorage.java:349) ~[hudi-aws-bundle-1.1.0.jar:1.1.0]
	at org.apache.hudi.storage.HoodieStorage.createImmutableFileInPath(HoodieStorage.java:312) ~[hudi-aws-bundle-1.1.0.jar:1.1.0]
	at org.apache.hudi.common.table.timeline.versioning.v2.ActiveTimelineV2.lambda$createCompleteFileInMetaPath$6(ActiveTimelineV2.java:758) ~[hudi-spark3.5-bundle_2.12-1.1.0.jar:1.1.0]
	at org.apache.hudi.common.table.timeline.SkewAdjustingTimeGenerator.consumeTime(SkewAdjustingTimeGenerator.java:63) ~[hudi-spark3.5-bundle_2.12-1.1.0.jar:1.1.0]
	at org.apache.hudi.common.table.timeline.versioning.v2.ActiveTimelineV2.createCompleteFileInMetaPath(ActiveTimelineV2.java:751) ~[hudi-spark3.5-bundle_2.12-1.1.0.jar:1.1.0]
	at org.apache.hudi.common.table.timeline.versioning.v2.ActiveTimelineV2.transitionStateToComplete(ActiveTimelineV2.java:566) ~[hudi-spark3.5-bundle_2.12-1.1.0.jar:1.1.0]
	at org.apache.hudi.common.table.timeline.versioning.v2.ActiveTimelineV2.saveAsComplete(ActiveTimelineV2.java:171) ~[hudi-spark3.5-bundle_2.12-1.1.0.jar:1.1.0]
	at org.apache.hudi.client.BaseHoodieWriteClient.commit(BaseHoodieWriteClient.java:321) ~[hudi-spark3.5-bundle_2.12-1.1.0.jar:1.1.0]
	at org.apache.hudi.client.BaseHoodieWriteClient.commitStats(BaseHoodieWriteClient.java:276) ~[hudi-spark3.5-bundle_2.12-1.1.0.jar:1.1.0]
	at org.apache.hudi.client.SparkRDDWriteClient.commit(SparkRDDWriteClient.java:149) ~[hudi-spark3.5-bundle_2.12-1.1.0.jar:1.1.0]
	at org.apache.hudi.HoodieSparkSqlWriterInternal.commitAndPerformPostOperations(HoodieSparkSqlWriter.scala:989) ~[hudi-spark3.5-bundle_2.12-1.1.0.jar:1.1.0]
	at org.apache.hudi.HoodieSparkSqlWriterInternal.writeInternal(HoodieSparkSqlWriter.scala:548) ~[hudi-spark3.5-bundle_2.12-1.1.0.jar:1.1.0]
	at org.apache.hudi.HoodieSparkSqlWriter$.write(HoodieSparkSqlWriter.scala:133) ~[hudi-spark3.5-bundle_2.12-1.1.0.jar:1.1.0]
	at org.apache.hudi.DefaultSource.createRelation(DefaultSource.scala:171) ~[hudi-spark3.5-bundle_2.12-1.1.0.jar:1.1.0]
	at org.apache.spark.sql.execution.datasources.SaveIntoDataSourceCommand.run(SaveIntoDataSourceCommand.scala:48) ~[spark-sql_2.12-3.5.6.jar:3.5.6]
	... Spark plumbing elided ...
	at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:244) ~[spark-sql_2.12-3.5.6.jar:3.5.6]
	at com.example.writer.HudiDataWriter.executeHudiWrite(HudiDataWriter.java:211) ~[app.jar:?]
	at com.example.writer.ParallelGroupProcessor.lambda$processGroups$0(ParallelGroupProcessor.java:127) ~[app.jar:?]
	at java.util.concurrent.CompletableFuture$AsyncRun.run(CompletableFuture.java:1804) [?:?]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) [?:?]
	at java.lang.Thread.run(Thread.java:840) [?:?]

2. Resulting file corruption

The completed deltacommit written by that failed commit, parsed as an Avro object container:

<instant>_<completion>.deltacommit          size = 3333 bytes    <-- corrupt
   avro.schema metadata: 3293 bytes; header ends at byte 3329
   block 0: records=1  declared_payload=13004  bytes_present=0
   *** TRUNCATED: expected total size 16353, actual 3333 ***

previous healthy deltacommit                 size = 17212 bytes
   block 0: records=1  declared_payload=13863  bytes_present=13879   OK

The file contains the complete Avro header plus exactly the 4-byte block header (record count = 1, payload length = 13004) and zero payload bytes. That is precisely the boundary inside DataFileWriter.writeBlock() between vout.flush() (which emits those 4 bytes) and buffer.writeTo(out) (the payload). The S3 ETag has no multipart suffix and there is a single object version, so this was one PUT of a truncated body — not a storage-layer fault.

Corroborating state after the failure:

• The metadata table's deltacommit for the same instant completed successfully ~558 ms earlier and is valid.
• All 37 log files for the instant were written intact.
• The marker directory .hoodie/.temp/<instant>/ was never cleaned, confirming commit finalization never ran.

3. Consequent permanent failure on every later operation

org.apache.hudi.exception.HoodieIOException: Failed to fetch HoodieCommitMetadata for instant
  ([<instant>__<completion>__deltacommit__COMPLETED])
	at org.apache.hudi.common.table.timeline.versioning.v2.ActiveTimelineV2.lambda$getCommitMetadataStream$4(ActiveTimelineV2.java:322)
	at org.apache.hudi.common.table.timeline.versioning.v2.ActiveTimelineV2.getLastCommitMetadataWithValidSchema(ActiveTimelineV2.java:299)
	at org.apache.hudi.common.table.TableSchemaResolver.getLatestCommitMetadataWithValidSchema(TableSchemaResolver.java:414)
	at org.apache.hudi.common.table.TableSchemaResolver.getTableSchemaFromLatestCommitMetadata(TableSchemaResolver.java:211)
	at org.apache.hudi.common.table.TableSchemaResolver.getTableAvroSchemaFromLatestCommit(TableSchemaResolver.java:286)
	at org.apache.hudi.HoodieSparkSqlWriterInternal.getLatestTableSchema(HoodieSparkSqlWriter.scala:679)
	at org.apache.hudi.HoodieSparkSqlWriterInternal.writeInternal(HoodieSparkSqlWriter.scala:361)
	...
Caused by: java.io.IOException: unable to read commit metadata for instant [...]
	at org.apache.hudi.common.table.timeline.versioning.v2.CommitMetadataSerDeV2.deserialize(CommitMetadataSerDeV2.java:88)
Caused by: java.lang.IllegalArgumentException: Could not deserialize metadata of type
  class org.apache.hudi.avro.model.HoodieCommitMetadata
	at org.apache.hudi.common.util.ValidationUtils.checkArgument(ValidationUtils.java:42)
	at org.apache.hudi.common.table.timeline.TimelineMetadataUtils.deserializeAvroMetadata(TimelineMetadataUtils.java:139)

Because getCommitMetadataStream() builds from getCommitsTimeline().filterCompletedInstants() and the lambda throws rather than skipping, the poisoned instant blocks schema resolution for every caller of TableSchemaResolver — writers, the compactor, the cleaner, and Spark reads. The exception is not an empty result, so the existing fallbacks to the table-config schema and to the latest data file are never reached.

Few Suggested fixes(Based on suggestions from Claude Opus Model):

These are independent and have different severities.

1. Make LocalRegistry.getCounter() incapable of returning null. Smallest, clearly correct fix:

private Counter getCounter(String name) {
    return counters.computeIfAbsent(name, k -> new Counter());
}

computeIfAbsent on the existing ConcurrentHashMap cannot return null, so a racing clear() costs a few counts instead of failing a commit. It also removes the synchronized on the hot path, which every FS write in the JVM currently contends on. (Alternatively, synchronize clear() on the same monitor — but computeIfAbsent is both correct and faster.)

2. Stop HoodieSparkSqlWriter.cleanup() from tearing down global metrics state. One table's save() calling Metrics.shutdownAllMetrics() destroys metrics for every other table in the JVM. This is wrong independent of the race — in a multi-table driver it also means per-table metrics are continuously clobbered and re-registered. Scoping the shutdown to the base path being written (or not shutting down at all on a normal save()) would fix both.

3. Never leave a partial commit-metadata object at a completed-instant path. This is the highest-severity item, because it converts any transient exception in that code path into permanent table corruption that no retry can clear. CommitMetadataSerDe.getInstantWriter streams Avro straight into the storage output stream, so a mid-serialization failure produces a file Hudi subsequently treats as authoritative. Options:

• Serialize to a byte array and issue a single write. Commit metadata is small (13 KB here); for tables with very large partition counts this trades memory, which is presumably why streaming was chosen, so it may need to be conditional.
• Or write to a temporary path and only move/copy to the final name after a successful close.
• Or verify the written object's length against the expected byte count immediately after close, and delete plus fail if it does not match.

Fixing (1) removes today's trigger. Fixing (3) removes the whole failure class, including future triggers.

4. Optional hardening: consider making getLastCommitMetadataWithValidSchema() skip unreadable instants rather than throwing, or at least surfacing a message that names the file and states that the table is unreadable until the instant is rolled back. Today the error gives no indication that the fix is to roll back the newest instant.

For now, as a mitigation step, we had deleted corrupted .delatcommit instant from timeline, deleted metadata table and reconstructed it. And, disabled hudi metrics hoodie.metrics.on=false. We had tried rollback but it didn't work as it required metadata table. And, since metadata table was inconsistent, we had only option to delete instant from timeline, and reconstruct metadata table.