KarmaGYZ commented on a change in pull request #15662:
URL: https://github.com/apache/flink/pull/15662#discussion_r615619196
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/metrics/ThresholdMeter.java
##########
@@ -33,7 +35,11 @@
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) {
Review comment:
It's better to doc this meter will not be used by any
performance-sensitive code paths.
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/metrics/ThresholdMeterTest.java
##########
@@ -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) {
Review comment:
```suggestion
private static Runnable repeat(Runnable task, int repeatNum) {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
