prashantwason commented on PR #19790:
URL: https://github.com/apache/hudi/pull/19790#issuecomment-6009732921
Pushed 2216888 and merged master (1c85053, conflicts with #19575 resolved).
Here is how `DistributedRegistry.register(jsc)` now works:
1. **Already registered with the current context** (`isRegistered()`):
return `this`.
2. **Never registered** (`registeredAppId == null`): register `this`, stamp
`registeredAppId`, return `this`. This is the normal first-time path. I reused
the `registeredAppId` field from #19575 as the "was registered before" marker
instead of adding a separate flag; it is still only stamped when this call
actually registers.
3. **Stale** (registered before, but its id is no longer in the current
`AccumulatorContext`, i.e. the previous SparkContext is gone and
`AccumulatorV2` refuses a second registration):
- If this instance was already replaced, delegate to the replacement
(`replacement = replacement.register(jsc)`). So every cache still holding the
original resolves to the same live accumulator, and a replacement that goes
stale after *another* restart is replaced in turn and swapped into
`REGISTRY_MAP` too (the map holds the replacement by then, not the original).
- Otherwise create a fresh `DistributedRegistry` that **shares** the
stale instance's `counters` map (no copy), register it, swap it into
`Registry.REGISTRY_MAP`, and remember it as `replacement`.
What changed versus the previous revision:
- **Counters are shared, not copied.** Previously the counters were copied
once at swap time, so anything still holding the stale instance (the metadata
writer's `registry` field, `HoodieWrapperFileSystem`'s static registries) kept
adding to an accumulator nobody reports, and those updates were lost. With a
shared map, those updates land in the counters the replacement reports.
Executors are unaffected: Spark ships `copy()`, which creates its own map; only
driver-side instances share.
- **Staleness is detected explicitly instead of via `catch
(IllegalStateException)`.** The broad catch also treated unrelated failures
(e.g. a stopped SparkContext) as staleness and then threw uncaught from the
second `register`, leaving the caches half-updated. Now any other registration
failure just propagates.
- **Tests:** `testRegisterHandlesStaleAccumulator` additionally checks that
an update through the stale instance is visible in the replacement, and that
after the replacement itself goes stale, a new registered replacement is
created, cached, keeps the counters, and is returned to holders of both older
instances.
`TestDistributedRegistry`, `TestDistributedRegistryUnderTaskRetry` and
`TestRecordIndexLookupMetrics` pass locally on the merged branch.
@danny0405 @rahil-c could you take another look when you get a chance?
Thanks!
--
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]
