[
https://issues.apache.org/jira/browse/FLINK-38783?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18127427#comment-18127427
]
abjfh edited comment on FLINK-38783 at 10/15/26 9:32 AM:
---------------------------------------------------------
PR #29488 fixes this issue and is ready for review:
https://github.com/apache/flink/pull/29488
[~tanyuxin] [~zakelly] [~martijnvisser] could one of you assign this ticket to
me (abjfh) and take a look? The previous PR for this ticket was closed by the
stale bot without a review.
was (Author: JIRAUSER309438):
PR #29488 fixes this issue and is ready for review:
[https://github.com/apache/flink/pull/29488]
Could a committer please assign this ticket to me (abjfh)? The previous PR for
this ticket was closed by the stale bot without a review.
> Hybrid shuffle: ConcurrentModificationException in
> TieredStorageResourceRegistry during concurrent result partition creation
> ----------------------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-38783
> URL: https://issues.apache.org/jira/browse/FLINK-38783
> Project: Flink
> Issue Type: Bug
> Components: Runtime / Network
> Affects Versions: 1.20.3
> Environment: flink 1.20.3
> hadoop 3.3.6
> Reporter: abjfh
> Priority: Critical
> Labels: bug, pull-request-available
>
> Problem
> With hybrid shuffle enabled (execution.batch-shuffle-mode), tasks
> occasionally fail while result partitions are being created. The failure is a
> ConcurrentModificationException thrown from
> TieredStorageResourceRegistry.registerResource. It occurs occasionally in
> production; the affected code is unchanged on master, and a multi-threaded
> reproduction is available (see Reproduction below).
> Stack trace (flink 1.20.3, hadoop 3.3.6)
> Caused by: java.util.ConcurrentModificationException
> at java.base/java.util.HashMap.computeIfAbsent(HashMap.java:1298) ~[?:?]
> at
> org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage.TieredStorageResourceRegistry.registerResource(TieredStorageResourceRegistry.java:46)
> ~[flink-dist-1.20.3.jar:1.20.3]
> Root cause
> TieredStorageResourceRegistry keeps its resources in a plain
> java.util.HashMap keyed by TieredStorageDataIdentifier, with a
> java.util.ArrayList of TieredStorageResource per key. A single registry
> instance is shared by the whole TaskManager: NettyShuffleServiceFactory
> creates it once and passes it to the tiered storage netty service and to the
> tier factories, so the result partitions of all task slots register into the
> same map.
> Registration runs while a result partition is created, on the task thread
> (ResultPartition.setup -> setupInternal) and on the Pekko actor thread
> (TaskExecutor.submitTask -> Task.<init> -> tier producer agent ->
> TieredStorageNettyServiceImpl.registerProducer). Removal runs when a
> partition is released, on the TaskManager ioExecutor
> (NettyShuffleEnvironment.releasePartition -> releasePartitionsLocally) and on
> the task canceller thread (ResultPartition.fail). Registration and removal
> therefore happen concurrently on different threads.
> registerResource() uses HashMap.computeIfAbsent(owner, ...).add(...).
> HashMap.computeIfAbsent verifies modCount after applying the mapping
> function, so any concurrent structural modification of the map makes it throw
> ConcurrentModificationException on the insert path - that is, on the first
> registration for a new result partition id, which matches the reported stack
> exactly. The same interleaving can also lose registrations (a registered
> resource is never released) or corrupt the table, which then shows up as
> NullPointerException or ArrayIndexOutOfBoundsException in clearResourceFor().
> Reproduction
> A new test (TieredStorageResourceRegistryTest) registers resources from 8
> threads with 5,000 registrations per thread and clears the owners afterwards:
> - same owner for all threads -> wrong release count, or NullPointerException
> in ArrayList.forEach
> - one fresh owner per registration -> ConcurrentModificationException from
> HashMap.computeIfAbsent, matching the reported trace
> Against the current code, 3 of the 5 test cases fail.
> Proposed fix
> Use ConcurrentHashMap instead of HashMap and CopyOnWriteArrayList instead of
> ArrayList, and register through ConcurrentHashMap.compute() so that adding a
> resource is linearized with the removal of the same owner. clearResourceFor()
> keeps removing the owner first and releasing the resources outside any map
> callback: releasing inside the callback would invert the lock order against
> TieredStorageNettyServiceImpl.registeredServiceProducers, whose
> computeIfAbsent callback registers here and whose release callback removes
> from that map.
> Verification
> With the fix, all 5 test cases pass and repeated runs are stable. The hybrid
> shuffle tiered storage suites (TieredResultPartitionTest,
> DiskTierProducerAgentTest, MemoryTierProducerAgentTest,
> RemoteTierProducerAgentTest, RemoteTierMasterAgentTest) pass as well,
> together with spotless, checkstyle and the ArchUnit architecture tests.
> Follow-up (separate issue)
> TieredStorageNettyServiceImpl.nettyConnectionReaderRegistrations is a plain
> HashMap as well. It is currently only accessed from one actor thread, so it
> is not hit by this bug, but nothing in the code enforces that invariant.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
