[
https://issues.apache.org/jira/browse/FLINK-41043?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18129238#comment-18129238
]
sepuri sai krishna commented on FLINK-41043:
--------------------------------------------
[~martijnvisser] could you assign this one to me?
I checked the concurrency on master at 77c3daea10d.
{{JobManagerSharedServices.fromConfiguration}} creates one {{ShuffleMaster}}
for the cluster, and {{DefaultJobMasterServiceFactory}} passes that same
instance to every JobMaster. {{JobMaster}} is a {{FencedRpcEndpoint}}, so each
job has its own main thread: {{startJobExecution}} calls
{{validateRunsInMainThread()}} and then {{shuffleMaster.registerJob}}, and
{{unregisterJob}} runs on that job's main thread executor. Concurrent jobs
therefore enter the same unsynchronized {{HashMap}} from different threads.
There is no synchronization anywhere in {{NettyShuffleMaster}}.
Besides the reported {{ClassCastException}}, a lost entry would also make
{{checkNotNull(jobShuffleContexts.get(jobId))}} fail in
{{getPartitionWithMetrics}} and in {{notifyPartitionRecoveryStarted}}.
One addition on scope. {{jobShuffleDescriptors}} is a plain {{HashMap}} as
well, and so is the per-job map inside it. It is written from
{{registerPartitionWithProducer}}, which also runs on the JobMaster main
thread, and from {{snapshotState}} and {{restoreState}}. Those writes sit
behind {{execution.batch.job-recovery.enabled}}, which defaults to false, so
that map is latent rather than active by default, but it is the same defect and
I would cover both in one change.
{{jobShuffleContexts}} is a plain {{HashMap}} with no synchronization on
release-1.20, release-2.2, release-2.3 and master, so this probably wants a
backport. Affects Version is currently 1.20.6 only.
> NettyShuffleMaster.registerJob fails with "HashMap$Node cannot be cast to
> HashMap$TreeNode" when jobs start concurrently
> ------------------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-41043
> URL: https://issues.apache.org/jira/browse/FLINK-41043
> Project: Flink
> Issue Type: Bug
> Components: Runtime / Coordination
> Affects Versions: 1.20.6
> Reporter: Martijn Visser
> Priority: Major
> Labels: test-stability
>
> https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=80031&view=logs&j=0c940707-2659-5648-cbe6-a1ad63045f0a
> (release-1.20 nightly 2026-10-11, test_ci table, Java 8)
> TimeFunctionsITCase failed 128 of its 350 cases. One JobMaster could not
> start:
> {code}
> [ERROR]
> org.apache.flink.table.planner.functions.TimeFunctionsITCase.test(TestCase)[301]
> -- Time elapsed: 0.492 s <<< ERROR!
> java.lang.RuntimeException: Failed to fetch next result
> ...
> Caused by: org.apache.flink.runtime.jobmaster.JobMasterException: Could not
> start the JobMaster.
> at
> org.apache.flink.runtime.jobmaster.JobMaster.onStart(JobMaster.java:462)
> ...
> Caused by: java.lang.ClassCastException: java.util.HashMap$Node cannot be
> cast to java.util.HashMap$TreeNode
> at
> org.apache.flink.runtime.shuffle.NettyShuffleMaster.registerJob(NettyShuffleMaster.java:192)
> at
> org.apache.flink.runtime.jobmaster.JobMaster.startJobExecution(JobMaster.java:1142)
> at
> org.apache.flink.runtime.jobmaster.JobMaster.onStart(JobMaster.java:460)
> {code}
> The other cases failed with "Job completed with illegal application status:
> UNKNOWN" (60) and "MiniCluster is not yet running or has already been shut
> down" (65).
> {{NettyShuffleMaster.jobShuffleContexts}} is a plain {{HashMap}} since
> FLINK-33985 (1.20.0). {{registerJob}} and {{unregisterJob}} are called from
> each JobMaster's main thread, and the ShuffleMaster is shared by all
> JobMasters through {{JobManagerSharedServices}}. {{BuiltInFunctionTestBase}}
> runs its cases concurrently against one MiniCluster. The field is the same on
> release-2.2, release-2.3 and master. First occurrence.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
