Retry requests on "Connection reset" and "Broken pipe" SocketException

Author: kaorimatz

src/main/java/com/treasuredata/client/TDRequestErrorHandler.java

@@ -182,8 +182,10 @@ else if (e instanceof SocketException) {
                 // All known SocketException are retryable.
                 return new TDClientSocketException(socketException);
             }
-            else if (Objects.equals(socketException.getMessage(), "Socket closed")) {
-                // okhttp can raise java.net.SocketException("Socket closed")
+            else if (Objects.equals(socketException.getMessage(), "Broken pipe") ||
+                    Objects.equals(socketException.getMessage(), "Connection reset") ||
+                    Objects.equals(socketException.getMessage(), "Socket closed")) {
+                // The underlying socket implementation used by OkHttp may throw these exceptions.
                 return new TDClientSocketException(socketException);
             }
             else {

src/test/java/com/treasuredata/client/TDRequestErrorHandlerTest.java

@@ -13,13 +13,16 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.net.SocketException;
 import java.time.Instant;
 import java.util.Date;
 import java.util.Optional;
 
 import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.is;
 import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 
 /**
  *
@@ -96,4 +99,35 @@ public void testParseRetryAfterSeconds()
         Instant expected = Instant.ofEpochMilli(now).plusSeconds(120);
         assertThat(retryAfter, is(expected));
     }
+
+    @Test
+    public void defaultExceptionResolverBrokenPipe()
+    {
+        SocketException socketException = new SocketException("Broken pipe");
+        TDClientException result = TDRequestErrorHandler.defaultExceptionResolver(socketException);
+        assertThat(result, instanceOf(TDClientSocketException.class));
+    }
+
+    @Test
+    public void defaultExceptionResolverConnectionReset()
+    {
+        SocketException socketException = new SocketException("Connection reset");
+        TDClientException result = TDRequestErrorHandler.defaultExceptionResolver(socketException);
+        assertThat(result, instanceOf(TDClientSocketException.class));
+    }
+
+    @Test
+    public void defaultExceptionResolverSocketClosed()
+    {
+        SocketException socketException = new SocketException("Socket closed");
+        TDClientException result = TDRequestErrorHandler.defaultExceptionResolver(socketException);
+        assertThat(result, instanceOf(TDClientSocketException.class));
+    }
+
+    @Test
+    public void defaultExceptionResolverUnknownSocketException()
+    {
+        SocketException socketException = new SocketException("Unknown socket error");
+        assertThrows(TDClientSocketException.class, () -> TDRequestErrorHandler.defaultExceptionResolver(socketException));
+    }
 }