KinesisIO throws ConcurrentModificationException on checkpoint

From time to time Kinesis IO throws ConcurrentModificationException on taking a checkpoint.

```

Caused by: java.util.ConcurrentModificationException
at java.util.ArrayDeque$DeqIterator.next(ArrayDeque.java:643)
at
org.apache.beam.sdks.java.io.kinesis.repackaged.com.google.common.collect.TransformedIterator.next(TransformedIterator.java:47)
at
org.apache.beam.sdks.java.io.kinesis.repackaged.com.google.common.collect.ImmutableCollection$Builder.addAll(ImmutableCollection.java:409)
at
org.apache.beam.sdks.java.io.kinesis.repackaged.com.google.common.collect.ImmutableList$Builder.addAll(ImmutableList.java:699)
at
org.apache.beam.sdks.java.io.kinesis.repackaged.com.google.common.collect.ImmutableList.copyOf(ImmutableList.java:256)
at
org.apache.beam.sdks.java.io.kinesis.repackaged.com.google.common.collect.ImmutableList.copyOf(ImmutableList.java:209)
at
org.apache.beam.sdk.io.kinesis.KinesisReaderCheckpoint.(KinesisReaderCheckpoint.java:44)
at
org.apache.beam.sdk.io.kinesis.KinesisReaderCheckpoint.asCurrentStateOf(KinesisReaderCheckpoint.java:49)
at
org.apache.beam.sdk.io.kinesis.KinesisReader.getCheckpointMark(KinesisReader.java:137)
at org.apache.beam.runners.flink.translation.wrappers.streaming.io.UnboundedSourceWrapper.snapshotState(UnboundedSourceWrapper.java:379)
at
org.apache.flink.streaming.api.functions.util.StreamingFunctionUtils.trySnapshotFunctionState(StreamingFunctionUtils.java:118)
at
org.apache.flink.streaming.api.functions.util.StreamingFunctionUtils.snapshotFunctionState(StreamingFunctionUtils.java:99)
at
org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.snapshotState(AbstractUdfStreamOperator.java:100)
at
org.apache.flink.streaming.api.operators.AbstractStreamOperator.snapshotState(AbstractStreamOperator.java:357)
...
11 more

```

*What is the issue*

org.apache.beam.sdk.io.kinesis.RoundRobin class is using ArrayDeque class which is not thread safe. If ConcurrentLinkedDeque deque is used the problem should be fixed.

*Beam 2.3 (master branch)*
Kinesis connector have been heavily refactored in master which makes me thinks how we should go about this fix? The org.apache.beam.sdk.io.kinesis.RoundRobin class doesn't exist in master anymore.

Imported from Jira [BEAM-2752](https://issues.apache.org/jira/browse/BEAM-2752). Original Jira may contain additional context.
Reported by: pawelbartoszek.

Start with KinesisReaderCheckpoint.java and KinesisReader.java at the stack-trace locations, then inspect the current Kinesis connector because the reported RoundRobin class no longer exists on master. Reproduce checkpointing under concurrent access and identify the current equivalent failure point; done means checkpoint creation no longer throws ConcurrentModificationException and has regression coverage.

Written by the indexing model from the issue text.