abjfh30 opened a new pull request, #29488:
URL: https://github.com/apache/flink/pull/29488
## What is the purpose of the change
Hybrid shuffle (`execution.batch-shuffle-mode`) creates all result
partitions of a TaskManager against one shared `TieredStorageResourceRegistry`.
Registration happens while a partition is created (task thread via
`ResultPartition.setup`, and the Pekko actor thread via
`TaskExecutor.submitTask` -> `Task.<init>` -> tier producer agent constructors
-> `TieredStorageNettyServiceImpl.registerProducer`), while removal happens
when a partition is released (TaskManager `ioExecutor` via
`NettyShuffleEnvironment.releasePartition`, and the task canceller thread via
`ResultPartition.fail`).
The registry used a plain `HashMap` and registered through
`HashMap.computeIfAbsent(owner, ...).add(...)`. `HashMap.computeIfAbsent`
verifies `modCount` after applying the mapping function, so a concurrent
structural modification of the map makes it throw
`ConcurrentModificationException` on the insert path - the failure reported in
FLINK-38783 - and the same interleaving can also silently lose registrations or
corrupt the table (`NullPointerException` / `ArrayIndexOutOfBoundsException`).
## Brief change log
- `TieredStorageResourceRegistry`: use `ConcurrentHashMap` instead of
`HashMap`, and a `CopyOnWriteArrayList` per identifier instead of an
`ArrayList`.
- Register through `ConcurrentHashMap.compute(...)` so that adding a
resource is linearized with the removal of the same owner, instead of
`computeIfAbsent(...).add(...)`, which can add a resource to a list that
`clearResourceFor()` has already removed from the map.
- `clearResourceFor()` stays unchanged: it removes the owner from the map
first and releases the resources outside of any map callback.
- Document in the class JavaDoc that the registry is shared by the whole
TaskManager and that registration and removal can run concurrently.
- Add `TieredStorageResourceRegistryTest` with 5 cases.
## Verifying this change
This change added tests and can be verified as follows:
- `TieredStorageResourceRegistryTest` (new): serial register/clear contract
(including re-registration and unknown-owner no-op); 8 threads x 5,000
registrations for the same owner; one fresh owner per registration; interleaved
registration/removal on shared owners, asserting that no resource is released
twice; 8 threads removing the same owner concurrently.
- Red/green: against the unmodified code, 3 of the 5 cases fail -
`ConcurrentModificationException` from `HashMap.computeIfAbsent` for the
fresh-owner shape (the exact stack of the report), and `NullPointerException`
in `ArrayList.forEach` plus wrong release counts for the shapes that share an
owner. With the change, all 5 pass, on repeated runs.
- Regression: `TieredResultPartitionTest`, `DiskTierProducerAgentTest`,
`MemoryTierProducerAgentTest`, `RemoteTierProducerAgentTest`,
`RemoteTierMasterAgentTest` (27 tests in total together with the new suite).
- `spotless:check`, `checkstyle:check` and the ArchUnit architecture tests
(`flink-architecture-tests-production`) pass.
## Relation to #27459 and why this is a separate pull request
#27459 targets the same root cause and was closed as stale without being
merged. This pull request is an independent, rebased take on the same bug:
- It registers through `ConcurrentHashMap.compute(...)` rather than
`computeIfAbsent(owner, ignore -> new CopyOnWriteArrayList<>()).add(...)`. Both
remove the `HashMap` race, but the `computeIfAbsent(...).add(...)` shape can
still add a resource to a list that a concurrent `clearResourceFor()` has
already removed from the map, in which case that resource is never released.
`compute(...)` linearizes the insertion with the removal of the same owner.
- `clearResourceFor()` deliberately keeps `release()` outside of the map
callback. Moving it inside (a natural simplification once the map is a
`ConcurrentHashMap`) would invert the lock order against
`TieredStorageNettyServiceImpl.registeredServiceProducers`: `registerProducer`
holds that map's bin lock and then registers here, while the release callback
registered here removes from that map, so that path would be an ABBA deadlock.
- The tests cover two distinct failure modes, because they need different
shapes to reproduce: the `ConcurrentModificationException` only appears on the
insert path (the first registration for a new identifier), while concurrent
registrations for an existing identifier show up as lost registrations instead.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with
`@Public(Evolving)`: no
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
(registration happens once per result partition, not per record)
- Anything that affects deployment or recovery: JobManager (and its
components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? no
- If yes, how is the feature documented? not applicable
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (please specify the tool below)
Generated-by: DeepSeek Harness (deepseek-v4-1-flash-260910)
--
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]
