[IMPROVEMENT] Fixing correctness and stability gaps for DistributedRegistry

## 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.)

## TL;DR — the 8 gaps at a glance

| # | Gap | Severity | Impacted scenarios |
|---|-----|----------|--------------------|
| 1 | Static `HoodieWrapperFileSystem.METRICS_REGISTRY_DATA/META` overwrites across tables — last write-client wins | Correctness (misattribution) | 2, 3, 4 |
| 2 | `DistributedRegistry` stays bound to a dead `SparkContext` after restart; new context's executor increments are lost | Correctness (silent metric loss) | 5 |
| 3 | Counters leak across write-client lifetimes — no clear on construct/close | Correctness (per-commit deltas lost) | 2, 4 |
| 4 | Executor-side `setRegistries` strips the table-name key; lookups by `(tableName, registryName)` silently miss | Latent footgun; becomes a bug once #1 lands | All |
| 5 | Task retries / speculation double-count (no `attemptNumber` dedup) | Correctness (bounded overcount) | 6 |
| 6 | Concurrent writers to the same table race on `setMetricsRegistry` | Correctness (torn writes) | 3 |
| 7 | `set(name, value)` is non-commutative — unsafe in an `AccumulatorV2` if ever called from an executor | Latent footgun | — |
| 8 | No flush between commits; counters grow monotonically in long-running apps | Operator UX (no per-commit observability) | 4 |

Each gap is detailed below with file:line references, severity, and a proposed fix.

## 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 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.

Start with DistributedRegistry.java, HoodieWrapperFileSystem.java, Registry.java, DistributedRegistryUtil.java, and SparkRDDWriteClient.java, tracing registry creation, registration, lookup, and lifecycle. Review the required tests for sequential and concurrent multi-tenant use, SparkContext restart, retries, and long-running applications. Done means the Phase 1 integration is correct and stable across the listed deployment scenarios without adding the deferred general framework features.

Written by the indexing model from the issue text.