[
https://issues.apache.org/jira/browse/FLINK-38783?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
huyuliang updated FLINK-38783:
------------------------------
Description:
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.
was:
Occasionally occurs, but no way to stably trigger it has been found.
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]
Summary: Hybrid shuffle: ConcurrentModificationException in
TieredStorageResourceRegistry during concurrent result partition creation
(was: Caused by: java.util.ConcurrentModificationException)
> 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
> 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)
