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 2019/08/01 13:25:42 UTC
[flink] 01/03: [FLINK-13326] Add closeInterruptibly()
This is an automated email from the ASF dual-hosted git repository.
sewen pushed a commit to branch release-1.9
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 1ec3a4c6239e15be742d585f42e63f4c9e060149
Author: Igal Shilman <ig...@data-artisans.com>
AuthorDate: Wed Jul 17 15:15:23 2019 +0200
[FLINK-13326] Add closeInterruptibly()
---
.../java/org/apache/flink/util/ResourceGuard.java | 39 ++++++++++++++++++----
1 file changed, 33 insertions(+), 6 deletions(-)
diff --git a/flink-core/src/main/java/org/apache/flink/util/ResourceGuard.java b/flink-core/src/main/java/org/apache/flink/util/ResourceGuard.java
index 83660a6..d27f8c8 100644
--- a/flink-core/src/main/java/org/apache/flink/util/ResourceGuard.java
+++ b/flink-core/src/main/java/org/apache/flink/util/ResourceGuard.java
@@ -94,13 +94,27 @@ public class ResourceGuard implements AutoCloseable, Serializable {
}
}
+ public void closeInterruptibly() throws InterruptedException {
+ synchronized (lock) {
+
+ closed = true;
+
+ while (leaseCount > 0) {
+ lock.wait();
+ }
+ }
+ }
+
/**
- * Closed the resource guard. This method will block until all calls to {@link #acquireResource()} have seen their
- * matching call to {@link #releaseResource()}.
+ * If the current thread is {@linkplain Thread#interrupt interrupted}
+ * while waiting for the close method to complete, then it will continue to wait.
+ * When the thread does return from this method its interrupt
+ * status will be set.
*/
- @Override
- public void close() {
+ @SuppressWarnings("WeakerAccess")
+ public void closeUninterruptibly() {
+ boolean interrupted = false;
synchronized (lock) {
closed = true;
@@ -109,11 +123,24 @@ public class ResourceGuard implements AutoCloseable, Serializable {
try {
lock.wait();
- } catch (InterruptedException ignore) {
- // Even on interruption, we cannot terminate the loop until all open leases are closed.
+ } catch (InterruptedException e) {
+ interrupted = true;
}
}
}
+
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ /**
+ * Closed the resource guard. This method will block until all calls to {@link #acquireResource()} have seen their
+ * matching call to {@link #releaseResource()}.
+ */
+ @Override
+ public void close() {
+ closeUninterruptibly();
}
/**