You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2018/03/09 18:58:37 UTC
[6/6] flink git commit: [hotfix] [network] [tests] Make
AwaitableBufferAvailablityListener thread-safe
[hotfix] [network] [tests] Make AwaitableBufferAvailablityListener thread-safe
This is called asynchronously by the spill writer and thus may need
synchronization on incrementing the counter but definately had visibility
issues with the counter. Using an AtomicLong fixes that.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/49623980
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/49623980
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/49623980
Branch: refs/heads/master
Commit: 496239806e68535c9c8291e320f8886d3a1b8709
Parents: 30eb8cd
Author: Nico Kruber <ni...@data-artisans.com>
Authored: Fri Feb 23 10:35:41 2018 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Fri Mar 9 16:49:40 2018 +0100
----------------------------------------------------------------------
.../partition/AwaitableBufferAvailablityListener.java | 12 +++++++-----
1 file changed, 7 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/49623980/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/AwaitableBufferAvailablityListener.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/AwaitableBufferAvailablityListener.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/AwaitableBufferAvailablityListener.java
index 2b6b834..6cf9d64 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/AwaitableBufferAvailablityListener.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/AwaitableBufferAvailablityListener.java
@@ -18,29 +18,31 @@
package org.apache.flink.runtime.io.network.partition;
+import java.util.concurrent.atomic.AtomicLong;
+
/**
* Test implementation of {@link BufferAvailabilityListener}.
*/
class AwaitableBufferAvailablityListener implements BufferAvailabilityListener {
- private long numNotifications;
+ private final AtomicLong numNotifications = new AtomicLong();
@Override
public void notifyDataAvailable() {
- ++numNotifications;
+ numNotifications.getAndIncrement();
}
public long getNumNotifications() {
- return numNotifications;
+ return numNotifications.get();
}
public void resetNotificationCounters() {
- numNotifications = 0;
+ numNotifications.set(0L);
}
void awaitNotifications(long awaitedNumNotifications, long timeoutMillis) throws InterruptedException {
long deadline = System.currentTimeMillis() + timeoutMillis;
- while (numNotifications < awaitedNumNotifications && System.currentTimeMillis() < deadline) {
+ while (numNotifications.get() < awaitedNumNotifications && System.currentTimeMillis() < deadline) {
Thread.sleep(1);
}
}