[FLINK-22340][runtime] Make ThresholdMeter thread safe.

flink-runtime/src/main/java/org/apache/flink/runtime/metrics/ThresholdMeter.java

@@ -23,6 +23,8 @@
 import org.apache.flink.util.clock.Clock;
 import org.apache.flink.util.clock.SystemClock;
 
+import javax.annotation.concurrent.GuardedBy;
+
 import java.time.Duration;
 import java.util.ArrayDeque;
 import java.util.Queue;
@@ -33,7 +35,11 @@ public class ThresholdMeter implements Meter {
     private final Clock clock;
     private final double maxEventsPerInterval;
     private final Duration interval;
+
+    @GuardedBy("this")
     private final Queue<Long> eventTimestamps;
+
+    @GuardedBy("this")
     private long eventCount = 0;
 
     public ThresholdMeter(double maxEventsPerInterval, Duration interval) {
@@ -52,13 +58,13 @@ public ThresholdMeter(double maxEventsPerInterval, Duration interval, Clock cloc
     }
 
     @Override
-    public void markEvent() {
+    public synchronized void markEvent() {
         eventTimestamps.add(clock.absoluteTimeMillis());
         eventCount++;
     }
 
     @Override
-    public void markEvent(long n) {
+    public synchronized void markEvent(long n) {
         long timestamp = clock.absoluteTimeMillis();
         for (int i = 0; i < n; i++) {
             eventTimestamps.add(timestamp);
@@ -72,20 +78,21 @@ public double getRate() {
     }
 
     @Override
-    public long getCount() {
+    public synchronized long getCount() {
         return eventCount;
     }
 
     public void checkAgainstThreshold() throws ThresholdExceedException {
-        if (getEventCountsRecentInterval() >= maxEventsPerInterval) {
+        int recentEvents = getEventCountsRecentInterval();
+        if (recentEvents >= maxEventsPerInterval) {
             throw new ThresholdExceedException(
                     String.format(
                             "%d events detected in the recent interval, reaching the threshold %f.",
-                            getEventCountsRecentInterval(), maxEventsPerInterval));
+                            recentEvents, maxEventsPerInterval));
         }
     }
 
-    private int getEventCountsRecentInterval() {
+    private synchronized int getEventCountsRecentInterval() {
         Long currentTimeStamp = clock.absoluteTimeMillis();
         while (!eventTimestamps.isEmpty()
                 && currentTimeStamp - eventTimestamps.peek() > interval.toMillis()) {

flink-runtime/src/test/java/org/apache/flink/runtime/metrics/ThresholdMeterTest.java

@@ -26,10 +26,13 @@
 import org.junit.Test;
 
 import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.TimeUnit;
 
 import static org.hamcrest.Matchers.closeTo;
 import static org.hamcrest.Matchers.is;
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.fail;
 
@@ -121,6 +124,49 @@ public void testUpdateInterval() {
         thresholdMeter.checkAgainstThreshold();
     }
 
+    @Test
+    public void testConcurrentAccess() throws Exception {
+        final ThresholdMeter thresholdMeter = new ThresholdMeter(THRESHOLD_LARGE, INTERVAL);
+        final int repeatNum = 100;
+        final int concurrency = 2;
+
+        final List<Thread> threads = new ArrayList<>();
+
+        threads.addAll(
+                getConcurrentThreads(repeat(thresholdMeter::markEvent, repeatNum), concurrency));
+        threads.addAll(
+                getConcurrentThreads(repeat(thresholdMeter::getRate, repeatNum), concurrency));
+        threads.addAll(
+                getConcurrentThreads(
+                        repeat(thresholdMeter::checkAgainstThreshold, repeatNum), concurrency));
+
+        for (Thread thread : threads) {
+            thread.start();
+        }
+
+        for (Thread thread : threads) {
+            thread.join();
+        }
+
+        assertEquals(repeatNum * concurrency, thresholdMeter.getCount());
+    }
+
+    private static Runnable repeat(Runnable task, int repeateNum) {
+        return () -> {
+            for (int i = 0; i < repeateNum; ++i) {
+                task.run();
+            }
+        };
+    }
+
+    private static List<Thread> getConcurrentThreads(Runnable task, int concurrency) {
+        List<Thread> threads = new ArrayList<>();
+        for (int i = 0; i < concurrency; ++i) {
+            threads.add(new Thread(task));
+        }
+        return threads;
+    }
+
     private static ThresholdMeter createLargeThresholdMeter() {
         return new ThresholdMeter(THRESHOLD_LARGE, INTERVAL, clock);
     }