[
https://issues.apache.org/jira/browse/FLINK-38783?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18131024#comment-18131024
]
abjfh commented on FLINK-38783:
-------------------------------
Update on the current state of the fix.
The implementation differs from the "Proposed fix" section above in one detail:
the final version keeps a plain ArrayList instead of switching to
CopyOnWriteArrayList. All list mutations now happen inside
ConcurrentHashMap.compute(), or inside the remove() of the owner, so a list is
never written after it has been removed from the map, and the bin lock of the
map provides the happens-before edge between a registration and the removal
that releases it. This also avoids copying the backing array on every
registration, and no longer implies that the list supports concurrent iteration.
I also tightened the interleaving test. The first version only asserted that a
resource is released at most once, so a registration that was silently lost,
and therefore never released, still passed. Every registration is now paired
with a removal of the same owner by the same thread, and the test asserts that
each resource is released exactly once.
The class javadoc documents the resulting contract: registration and removal
are linearized per owner, a resource is released at most once, and a resource
registered after the last removal of its owner is never released, so callers
must register everything before that removal. The tiered storage lifecycle
satisfies this, since resources are registered while the result partition is
created and a partition can only be released after it has been set up.
With this, all 5 cases of TieredStorageResourceRegistryTest pass and repeated
runs are stable. TieredResultPartitionTest, DiskTierProducerAgentTest,
MemoryTierProducerAgentTest, RemoteTierProducerAgentTest and
RemoteTierMasterAgentTest pass as well, together with spotless, checkstyle and
the ArchUnit tests.
The PR is #29488 and is ready for review. The follow-up mentioned above,
TieredStorageNettyServiceImpl.nettyConnectionReaderRegistrations, is untouched
and still needs its own issue.
> 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
(v10.3.26#10030026)
