[SPARK-3154][STREAMING] Replace ConcurrentHashMap with mutable.HashMap and remove @volatile from 'stopped'

[zsxwing]

Since sequenceNumberToProcessor and stopped are both protected by the lock sequenceNumberToProcessor, ConcurrentHashMap and volatile is unnecessary. So this PR updated them accordingly.

[SparkQA]

Test build #24220 has started for PR 3634 at commit 0d087ac.

• This patch merges cleanly.

[SparkQA]

Test build #24220 has finished for PR 3634 at commit 0d087ac.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[AmplabJenkins]

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/24220/
Test PASSed.

[srowen]

LGTM

[harishreedharan]

+1. Looks good!

[aarondav]

Could use the @GuardedBy("sequenceNumberToProcessor") javax annotation.

[zsxwing]

external/flume-sink doesn't depend on jsr305. GuardedBy is in jsr305***.jar. However, great to know that we can use GuardedBy in Spark.

[aarondav]

Ah, did not realize it was not a javax standard. Thanks.

[aarondav]

LGTM too, at your discretion you could replace the comment with the annotation or not. Will merge when addressed.

[aarondav]

Merging into master, thanks!