[FLINK-32335] Fix the ConcurrentModificationException in the unittest

This closes apache#247.
lindong28 · Jul 14, 2023 · 892fb47 · 892fb47
1 parent fbd2a52
commit 892fb47
Showing 1 changed file with 8 additions and 6 deletions.
diff --git a/...k-ml-iteration-common/src/main/java/org/apache/flink/iteration/operator/HeadOperator.java b/...k-ml-iteration-common/src/main/java/org/apache/flink/iteration/operator/HeadOperator.java
@@ -461,12 +461,14 @@ private List<AbstractEvent> parseInputChannelEvents(InputChannel inputChannel)
             PrioritizedDeque<BufferConsumerWithPartialRecordLength> queue =
                     ReflectionUtils.getFieldValue(
                             pipelinedSubpartition, PipelinedSubpartition.class, "buffers");
-            for (BufferConsumerWithPartialRecordLength bufferConsumer : queue) {
-                if (!bufferConsumer.getBufferConsumer().isBuffer()) {
-                    events.add(
-                            EventSerializer.fromBuffer(
-                                    bufferConsumer.getBufferConsumer().copy().build(),
-                                    getClass().getClassLoader()));
+            synchronized (queue) {
+                for (BufferConsumerWithPartialRecordLength bufferConsumer : queue) {
+                    if (!bufferConsumer.getBufferConsumer().isBuffer()) {
+                        events.add(
+                                EventSerializer.fromBuffer(
+                                        bufferConsumer.getBufferConsumer().copy().build(),
+                                        getClass().getClassLoader()));
+                    }
                 }
             }
         } else {