nsivabalan opened a new issue, #19063:
URL: https://github.com/apache/hudi/issues/19063
## Background
`DistributedRegistry`
(`hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/metrics/DistributedRegistry.java`)
is a `Spark AccumulatorV2` subclass that aggregates `String → long` counters
incremented on executors back to the driver. Today its only production consumer
is `HoodieWrapperFileSystem`, which uses two registries
(`HoodieWrapperFileSystem` for data-file paths and
`HoodieWrapperFileSystemMetaFolder` for `.hoodie/` paths) to record per-FS-call
counts, durations, and bytes for `create / rename / delete / listStatus /
mkdirs / getFileStatus / globStatus / listFiles / read / write`.
Wiring:
- Created in
`DistributedRegistryUtil.createWrapperFileSystemRegistries(...)`
(`hudi-client/hudi-client-common/.../util/DistributedRegistryUtil.java:39`)
- Invoked from `SparkRDDWriteClient` constructor
(`hudi-client/hudi-spark-client/.../SparkRDDWriteClient.java:81`)
- Gated by `hoodie.metrics.on=true` AND `hoodie.metrics.executor.enable=true`
- Stashed in process-wide static maps: `Registry.REGISTRY_MAP`
(`hudi-io/.../common/metrics/Registry.java:51`) and
`HoodieSparkEngineContext.DISTRIBUTED_REGISTRY_MAP`
(`hudi-client/hudi-spark-client/.../client/common/HoodieSparkEngineContext.java:96`)
- Made reachable on executors via static fields on `HoodieWrapperFileSystem`
(`hudi-hadoop-common/.../HoodieWrapperFileSystem.java:101-107`)
## Scope of this issue (Phase 1)
**This issue is intentionally narrow.** Phase 1 scope is *strictly limited*
to making the existing `HoodieWrapperFileSystem` integration correct and stable
across all deployment scenarios. We are **not** trying to generalize
`DistributedRegistry` into a broader executor-side telemetry framework as part
of this work.
That said, the underlying machinery is potentially useful well beyond
`HoodieWrapperFileSystem` (e.g. instrumenting index lookup paths,
metadata-table reads, log-merge internals, etc.). Those use cases require
additional capabilities — histogram / distribution support, a clean
executor-side lookup API, broader instrumentation rollout — and are
**explicitly deferred to follow-up issues**. Several of the fixes in this issue
(notably #4) lay groundwork for those follow-ups but stop short of building
them out.
Out of scope for this issue:
- Adding histogram / distribution support to `Registry`
- Generalizing the framework to other instrumentation points (index lookup,
MDT reads, etc.)
- Refactoring `Registry.REGISTRY_MAP` away from being process-wide static
Those will be tracked separately.
## Deployment scenarios the integration must support
1. **Single-tenant batch:** one `SparkSession`, one write to one table,
exits.
2. **Multi-tenant sequential:** one `SparkSession` writes to multiple tables
in sequence in the same JVM.
3. **Multi-tenant concurrent:** one `SparkSession`, multiple threads, each
writing to a different table concurrently.
4. **Long-running / streaming:** a single `SparkSession` that lives for days
and processes many micro-batches (Structured Streaming, long-running ingestion
services).
5. **SparkContext restart:** `SparkContext` is stopped and a new one is
started in the same JVM (Spark shells, notebooks, Spark Connect).
6. **Task retries / speculative execution.**
7. **Executor death** and Spark relaunching tasks on a new executor.
(Spark's own accumulator machinery handles this — calling it out for
completeness.)
## Bugs / gaps to fix
### 1. Static `HoodieWrapperFileSystem.METRICS_REGISTRY_DATA/META`
overwrites across tables
```java
// HoodieWrapperFileSystem.java:101-107
private static Registry METRICS_REGISTRY_DATA;
private static Registry METRICS_REGISTRY_META;
public static void setMetricsRegistry(Registry registry, Registry
registryMeta) {
METRICS_REGISTRY_DATA = registry;
METRICS_REGISTRY_META = registryMeta;
}
```
The fields are static and `setMetricsRegistry` blindly overwrites. In
scenarios 2/3/4, the last `SparkRDDWriteClient` constructed wins, and all
subsequent FS operations attribute to its registry regardless of which table
the path actually belongs to.
**Severity:** correctness — metrics are silently misattributed to the wrong
table.
**Proposed fix:** route by path. Replace the static `Registry` fields with a
registry lookup keyed by table base path (which is derivable from the `Path`
already passed to the wrapper). On `setMetricsRegistry`, insert into the map
keyed by the calling table's base path. `getMetricRegistryForPath(Path)` then
matches the longest table base path prefix.
### 2. Accumulator bound to a dead `SparkContext`
`DistributedRegistry.register(jsc)` (line 48-52):
```java
public void register(JavaSparkContext jsc) {
if (!isRegistered()) {
jsc.sc().register(this);
}
}
```
Combined with `REGISTRY_MAP` being a process-wide static that is never
cleared, scenario 5 breaks: a new `SparkContext` in the same JVM finds the
existing `DistributedRegistry`, `isRegistered()` returns `true` against the
dead `SparkContext`, and `register()` silently no-ops. Subsequent executor-side
increments have nowhere to merge to on the new driver.
**Severity:** correctness — metrics silently stop flowing after a
`SparkContext` restart.
**Proposed fix:** when `getMetricRegistry(...)` is called, detect that the
stored registry is bound to a different `SparkContext` than the current one,
evict it from both static maps, and create a fresh `DistributedRegistry`
against the new context.
### 3. Counters leak across write-client lifetimes
`SparkRDDWriteClient` constructor calls
`createWrapperFileSystemRegistries(...)` which does `computeIfAbsent` on the
static map. If a previous write client to the same table already populated the
registry, the new client inherits those counts. No `close()` path clears them.
Scenarios 2 and 4 hit this — sequential writes to the same table accumulate
cross-write counts, and a streaming app grows the registry monotonically across
micro-batches.
**Severity:** correctness — per-commit metric deltas are unrecoverable.
**Proposed fix:** clear the registry (`registry.reset()`) on write-client
construction *and* on `close()`. Construction-side clear handles crash
recovery; close-side clear handles clean lifecycle.
### 4. Executor-side registry lookup loses the table-name key
```java
// Registry.java:152-156
static void setRegistries(Collection<Registry> registries) {
for (Registry registry : registries) {
REGISTRY_MAP.putIfAbsent(makeKey("", registry.getName()), registry);
}
}
```
The driver-side key is `tableName::registryName`, but `setRegistries`
re-keys to `""::tableName.registryName` on the executor. Any caller doing
`Registry.getRegistryOfClass(tableName, registryName, ...)` on the executor
will miss and create a fresh `LocalRegistry` that nobody reads.
Today this is latent because `HoodieWrapperFileSystem` uses its own
static-field path (`setMetricsRegistry`) rather than the lookup API. But if fix
#1 moves to a per-table-keyed lookup, this becomes load-bearing. (It is also
what would block any future generalization to other call sites, so fixing it
now pays forward.)
**Severity:** latent footgun; becomes a correctness bug once #1 is fixed.
**Proposed fix:** preserve the full `(tableName, registryName)` key on the
executor side. Either ship the original keys alongside the registries in
`setRegistries`, or have `Registry.getName()` encode the table name and have
the executor re-parse.
### 5. Task retries and speculative execution double-count
`AccumulatorV2` delivers at-least-once semantics. If a task increments 100
counters, fails, and retries, the driver merges both attempts and sees 200.
Same for speculative duplicates. There is no `attemptNumber`-based dedup.
For FS-op counters specifically, this means "my Hudi-reported FS op count
doesn't match my cloud provider's bill" — a real operator complaint.
**Severity:** correctness — bounded overcounting under retries / speculation.
**Proposed fix options (in order of effort):**
- (a) Document the at-least-once semantic and accept it. Cheapest and
defensible — Spark accumulators have always been this way.
- (b) Buffer in a task-local map, emit to the accumulator only from a
`TaskCompletionListener` that fires on success. Correct, but adds plumbing.
Suggest shipping (a) by default and exposing (b) behind an opt-in config for
users who need exact counts.
### 6. Concurrent writers to the same table
Scenario 3 — two threads construct `SparkRDDWriteClient` for the same table.
`computeIfAbsent` makes registry *creation* race-safe, but `setMetricsRegistry`
is racy with concurrent increments from in-flight Spark jobs.
**Severity:** correctness — torn writes to the static field.
**Proposed fix:** falls out of #1 (per-table-keyed lookup makes the writes
target different map entries). If #1 isn't taken, at minimum
`setMetricsRegistry` needs synchronization.
### 7. `set(name, value)` is unsafe inside an AccumulatorV2
```java
// DistributedRegistry.java:70-72
public void set(String name, long value) {
counters.merge(name, value, (oldValue, newValue) -> newValue);
}
```
Accumulators must be commutative and associative. `set` is neither — the
result depends on merge order. `HoodieWrapperFileSystem` doesn't call `set`, so
this is latent today, but it's a footgun.
**Severity:** latent; would produce non-deterministic results if called from
an executor.
**Proposed fix:** either remove `set` from `DistributedRegistry`, or throw
if invoked when `TaskContext.get() != null` (i.e. on an executor).
### 8. No flush between commits
`registry.clear()` is only called from `Metrics.getAllMetrics(flush=true)`,
which fires on `Metrics.flush()` / shutdown — not per commit. In a long-running
streaming app, FS counters grow unboundedly across micro-batches. Per-commit
deltas are not derivable from the registry alone.
**Severity:** operator UX — per-commit observability is lost.
**Proposed fix:** snapshot-and-clear at the end of each `commit()`. Timing
matters — the reporter must scrape before the clear, or the values are lost.
Easiest path: integrate with the existing `getAllMetrics(flush=true)` call from
the reporter chain.
## Suggested PR breakdown
Splitting into three PRs to keep each focused and reviewable:
- **PR 1 (must-fix, correctness):** #1 + #3 + #2. Without these, metrics are
misattributed in multi-tenant and long-running scenarios, which is worse than
not having them.
- **PR 2 (correctness polish):** #4 + #7 + #8.
- **PR 3 (retry story):** #5, with the opt-in `TaskCompletionListener` mode
behind a config.
#6 falls out of PR 1.
## Tests required
Per scenario:
- **Multi-tenant sequential** (#1, #3): in one JVM, write to table A then
table B, assert each registry contains only that table's FS ops.
- **Multi-tenant concurrent** (#1, #6): two threads, two tables, two
concurrent Spark jobs, assert no cross-attribution.
- **Long-running** (#3, #8): same write-client / engine context, run 10
commits, assert per-commit deltas are recoverable and total state doesn't grow
monotonically across commits.
- **SparkContext restart** (#2): stop and recreate the context in the same
JVM, assert the new context's executor increments land on the new driver.
- **Task retry** (#5): inject a task failure on first attempt, assert the
documented semantic (over-count under default mode; exact count under opt-in
mode).
- **End-to-end functional**: a real `bulk_insert` / `upsert` against a local
filesystem with executor metrics enabled, asserting non-zero counter values for
`create`, `getFileStatus`, etc. on the driver after job completion. This is
missing today.
## Follow-up work (separate issues)
Once Phase 1 lands and `DistributedRegistry` is solid for its current use
case, natural follow-ups include:
- Histogram / distribution primitives in `Registry`, so executor-side
latency can be reported with p50/p95/p99 rather than just count + sum.
- Broader instrumentation rollout (index lookup paths, metadata-table read
paths, log-merge internals) using the now-trustworthy framework.
- Lifecycle refactor — tying registry lifetime to `HoodieEngineContext` /
write client rather than the JVM process — to eliminate the remaining
static-state concerns.
These will be filed as separate issues. Phase 1 deliberately stops at the
boundary of "current `HoodieWrapperFileSystem` integration works correctly,"
because that scope is shippable on its own and unblocks operators today.
--
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]
