Merge pull request #54 from rethinkdb/release-candidate/2.4.3-hotfixes

Release with hotfixes
rethinkdb · Jun 4, 2020 · 01b17f8 · 01b17f8
2 parents a589e79 + ddfde38
commit 01b17f8
Show file tree

Hide file tree

Showing 4 changed files with 303 additions and 185 deletions.
diff --git a/build.gradle.kts b/build.gradle.kts
@@ -13,7 +13,7 @@ plugins {
     id("com.jfrog.bintray") version "1.8.4"
 }
 
-version = "2.4.3"
+version = "2.4.4"
 group = "com.rethinkdb"
 
 java.sourceCompatibility = JavaVersion.VERSION_1_8

diff --git a/src/main/java/com/rethinkdb/net/DefaultConnectionFactory.java b/src/main/java/com/rethinkdb/net/DefaultConnectionFactory.java
@@ -18,6 +18,7 @@
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicReference;
 
 /**
  * The default {@link ConnectionSocket.Factory} and {@link ResponsePump.Factory} for any default connections.
@@ -177,6 +178,7 @@ public String toString() {
     }
 
     private static class ThreadResponsePump implements ResponsePump {
+        private final AtomicReference<Throwable> shutdownReason = new AtomicReference<>();
         private final Thread thread;
         private Map<Long, CompletableFuture<Response>> awaiting = new ConcurrentHashMap<>();
 
@@ -220,7 +222,7 @@ public ThreadResponsePump(ConnectionSocket socket, boolean daemon) {
         @Override
         public @NotNull CompletableFuture<Response> await(long token) {
             if (awaiting == null) {
-                throw new ReqlDriverError("Response pump closed.");
+                throw new ReqlDriverError("Response pump closed.", shutdownReason.get());
             }
             CompletableFuture<Response> future = new CompletableFuture<>();
             awaiting.put(token, future);
@@ -234,6 +236,7 @@ public boolean isAlive() {
 
         private void shutdown(Throwable t) {
             Map<Long, CompletableFuture<Response>> awaiting = this.awaiting;
+            this.shutdownReason.compareAndSet(null, t);
             this.awaiting = null;
             thread.interrupt();
             if (awaiting != null) {
@@ -243,7 +246,7 @@ private void shutdown(Throwable t) {
 
         @Override
         public void shutdownPump() {
-            shutdown(new ReqlDriverError("Response pump closed."));
+            shutdown(new Throwable("Shutdown was requested."));
         }
 
         @Override