Potential ConcurrentModificationException working with Segment attributes

[RaulGracia]

Describe the bug
In a recent longevity experiment, we got this exception from the Segment Store:

io.pravega.segmentstore.storage.metadata.StorageMetadataException: Transaction failed
       at io.pravega.segmentstore.storage.metadata.TableBasedMetadataStore.handleException(TableBasedMetadataStore.java:201)
       at io.pravega.segmentstore.storage.metadata.TableBasedMetadataStore.lambda$writeAll$8(TableBasedMetadataStore.java:189)
       at java.base/java.util.concurrent.CompletableFuture.uniExceptionally(Unknown Source)
       at java.base/java.util.concurrent.CompletableFuture$UniExceptionally.tryFire(Unknown Source)
       at java.base/java.util.concurrent.CompletableFuture.postComplete(Unknown Source)
       at java.base/java.util.concurrent.CompletableFuture.completeExceptionally(Unknown Source)
       at io.pravega.common.function.Callbacks.invokeSafely(Callbacks.java:54)
       at io.pravega.common.concurrent.Futures.lambda$exceptionListener$5(Futures.java:277)
       at java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(Unknown Source)
       at java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(Unknown Source)
       at java.base/java.util.concurrent.CompletableFuture.postComplete(Unknown Source)
       at java.base/java.util.concurrent.CompletableFuture.completeExceptionally(Unknown Source)
       at io.pravega.common.function.Callbacks.invokeSafely(Callbacks.java:54)
       at io.pravega.segmentstore.server.logs.operations.CompletableOperation.fail(CompletableOperation.java:111)
       at io.pravega.segmentstore.server.logs.OperationProcessor$QueueProcessingState.failOperation(OperationProcessor.java:672)
       at io.pravega.segmentstore.server.logs.OperationProcessor.processOperations(OperationProcessor.java:318)
       at java.base/java.util.concurrent.CompletableFuture$UniAccept.tryFire(Unknown Source)
       at java.base/java.util.concurrent.CompletableFuture$Completion.run(Unknown Source)
       at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
       at java.base/java.util.concurrent.FutureTask.run(Unknown Source)
       at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(Unknown Source)
       at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
       at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
       at java.base/java.lang.Thread.run(Unknown Source)
Caused by: java.util.ConcurrentModificationException: null
       at java.base/java.util.HashMap$HashIterator.nextNode(Unknown Source)
       at java.base/java.util.HashMap$EntryIterator.next(Unknown Source)
       at java.base/java.util.HashMap$EntryIterator.next(Unknown Source)
       at io.pravega.common.util.CollectionHelpers$ConvertedIterator.next(CollectionHelpers.java:271)
       at com.google.common.collect.Iterators$ConcatenatedIterator.next(Iterators.java:1364)
       at com.google.common.collect.Iterators$1.next(Iterators.java:141)
       at java.base/java.util.HashMap.putMapEntries(Unknown Source)
       at java.base/java.util.HashMap.<init>(Unknown Source)
       at io.pravega.segmentstore.server.logs.SegmentMetadataUpdateTransaction.getAttributes(SegmentMetadataUpdateTransaction.java:148)

The problem is detected SegmentMetadataUpdateTransaction, where in its getAttributes it does:

public Map<UUID, Long> getAttributes() {
        HashMap<UUID, Long> result = new HashMap<>(this.baseAttributeValues); // Problem happens here
        result.putAll(this.attributeUpdates);
        return result;
}

The previous HashMap constructor iterates over the input Map entries to populate the new one. This means that, somewhere else and at the same time, the baseAttributeValues map is being modified. While this class does not modify that Map, we see that it is actually initialized in the constructor as a reference for baseMetadata.getAttributes().

It turns out that the actual object is of a StreamSegmentMetadata.AttributesView object. While that class is meant to be synchronized and non-modifiable, we also see that the implementation of entrySet(), used in the HashMap constructor, is like this:

        public Set<Map.Entry<UUID, Long>> entrySet() {
            synchronized (StreamSegmentMetadata.this) {
                return CollectionHelpers.joinSets(
                        coreAttributes.entrySet(), Callbacks::identity,
                        extendedAttributes.entrySet(), e -> new AbstractMap.SimpleImmutableEntry<>(e.getKey(), e.getValue().value));
            }
        }

So, what it is used to any class using a reference to this object is whather is returned from CollectionHelpers.joinSets, which in turn says that:

Returns an unmodifiable Set View made up of the given Sets while translating the items into a common type. The returned
Set View does not copy any of the data from any of the given Sets, therefore any changes in the two Sets will be
reflected in the View.

Therefore, what other classes will use when calling to entrySet() is not a copy, but a reference to the coreAttributes and extendedAttributes maps in StreamSegmentMetadata. As these maps can be modified along a Segment's lifetime, if any class is iterating the StreamSegmentMetadata.AttributesView at the same time that coreAttributes or extendedAttributes are modified, could get a ConcurrentModificationException.

To Reproduce
Hard to reproduce. It would require a concurrent iteration and modification of a Segment's attributes.

Screenshots
n/a

Additional information
Using ConcurrentHashMap for Segment attributes may prevent this problem from happening.