Issue 1544: ConcurrentModificationException with nonblocking logReade…

…r.readNext(true)

*Motivation*

Fixes apache#1544. ConcurrentModificationException is thrown when trying to access a non-thread-safe hashmap from different threads.

*Changes*

Make sure accessing to this non-thread-safe hashmap is under synchronized block.

*Tests*

It is a bit tricky to reproduce this race condition in a unit test or an integration test. so not going to attempt adding any tests.

stream/distributedlog/core/src/main/java/org/apache/distributedlog/impl/ZKLogSegmentMetadataStore.java

@@ -476,8 +476,12 @@ void notifyLogStreamDeleted(String logSegmentsPath,
             return;
         }
         this.submitTask(logSegmentsPath, () -> {
-            for (LogSegmentNamesListener listener : listeners.keySet()) {
-                listener.onLogStreamDeleted();
+            // the listener map might be updated in different threads (e.g. unregisterLogSegmentListener)
+            // so access it under a synchronization block
+            synchronized (listeners) {
+                for (LogSegmentNamesListener listener : listeners.keySet()) {
+                    listener.onLogStreamDeleted();
+                }
             }
         });
 
@@ -490,8 +494,12 @@ void notifyLogSegmentsUpdated(String logSegmentsPath,
             return;
         }
         this.submitTask(logSegmentsPath, () -> {
-            for (VersionedLogSegmentNamesListener listener : listeners.values()) {
-                listener.onSegmentsUpdated(segments);
+            // the listener map might be updated in different threads (e.g. unregisterLogSegmentListener)
+            // so access it under a synchronization block
+            synchronized (listeners) {
+                for (VersionedLogSegmentNamesListener listener : listeners.values()) {
+                    listener.onSegmentsUpdated(segments);
+                }
             }
         });
     }