bhavya-ganatra opened a new issue, #19570:
URL: https://github.com/apache/hudi/issues/19570
### 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:
```java
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.
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
