[
https://issues.apache.org/jira/browse/FLINK-38783?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
huyuliang updated FLINK-38783:
------------------------------
Component/s: Runtime / Network
> 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: huyuliang
> 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]
> at
> org.apache.flink.runtime.io.network.partition.hybrid.tiered.netty.TieredStorageNettyServiceImpl.lambda$registerProducer$1(TieredStorageNettyServiceImpl.java:80)
> ~[flink-dist-1.20.3.jar:1.20.3]
> at
> java.base/java.util.concurrent.ConcurrentHashMap.computeIfAbsent(ConcurrentHashMap.java:1737)
> ~[?:?]
> at
> org.apache.flink.runtime.io.network.partition.hybrid.tiered.netty.TieredStorageNettyServiceImpl.registerProducer(TieredStorageNettyServiceImpl.java:76)
> ~[flink-dist-1.20.3.jar:1.20.3]
> at
> org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.memory.MemoryTierProducerAgent.<init>(MemoryTierProducerAgent.java:100)
> ~[flink-dist-1.20.3.jar:1.20.3]
> at
> org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.memory.MemoryTierFactory.createProducerAgent(MemoryTierFactory.java:114)
> ~[flink-dist-1.20.3.jar:1.20.3]
> at
> org.apache.flink.runtime.io.network.partition.hybrid.tiered.shuffle.TieredResultPartitionFactory.createTierProducerAgentsAndMemorySpecs(TieredResultPartitionFactory.java:227)
> ~[flink-dist-1.20.3.jar:1.20.3]
> at
> org.apache.flink.runtime.io.network.partition.hybrid.tiered.shuffle.TieredResultPartitionFactory.createTieredResultPartition(TieredResultPartitionFactory.java:123)
> ~[flink-dist-1.20.3.jar:1.20.3]
> at
> org.apache.flink.runtime.io.network.partition.ResultPartitionFactory.create(ResultPartitionFactory.java:260)
> ~[flink-dist-1.20.3.jar:1.20.3]
> at
> org.apache.flink.runtime.io.network.partition.ResultPartitionFactory.create(ResultPartitionFactory.java:145)
> ~[flink-dist-1.20.3.jar:1.20.3]
> at
> org.apache.flink.runtime.io.network.NettyShuffleEnvironment.createResultPartitionWriters(NettyShuffleEnvironment.java:238)
> ~[flink-dist-1.20.3.jar:1.20.3]
> at org.apache.flink.runtime.taskmanager.Task.<init>(Task.java:416)
> ~[flink-dist-1.20.3.jar:1.20.3]
> at
> org.apache.flink.runtime.taskexecutor.TaskExecutor.submitTask(TaskExecutor.java:858)
> ~[flink-dist-1.20.3.jar:1.20.3]
> at
> java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native
> Method) ~[?:?]
> at
> java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
> ~[?:?]
> at
> java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
> ~[?:?]
> at java.base/java.lang.reflect.Method.invoke(Method.java:566) ~[?:?]
> at
> org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.lambda$handleRpcInvocation$1(PekkoRpcActor.java:318)
> ~[flink-rpc-akka940ac193-56ac-4139-bf57-df841d42fabd.jar:1.20.3]
> at
> org.apache.flink.runtime.concurrent.ClassLoadingUtils.runWithContextClassLoader(ClassLoadingUtils.java:83)
> ~[flink-dist-1.20.3.jar:1.20.3]
> at
> org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.handleRpcInvocation(PekkoRpcActor.java:316)
> ~[flink-rpc-akka940ac193-56ac-4139-bf57-df841d42fabd.jar:1.20.3]
> at
> org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.handleRpcMessage(PekkoRpcActor.java:229)
> ~[flink-rpc-akka940ac193-56ac-4139-bf57-df841d42fabd.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)
