From 3580f4db8166cd2625a3a796f044bb173ef00391 Mon Sep 17 00:00:00 2001
From: Satoshi Matsumoto <kaorimatz@gmail.com>
Date: Wed, 11 Jun 2025 18:26:00 +0900
Subject: [PATCH] Retry requests on "Connection reset" and "Broken pipe"
 SocketException

---
 .../client/TDRequestErrorHandler.java         |  6 ++--
 .../client/TDRequestErrorHandlerTest.java     | 34 +++++++++++++++++++
 2 files changed, 38 insertions(+), 2 deletions(-)

diff --git a/src/main/java/com/treasuredata/client/TDRequestErrorHandler.java b/src/main/java/com/treasuredata/client/TDRequestErrorHandler.java
index 5c780e52..1f2d0fd8 100644
--- a/src/main/java/com/treasuredata/client/TDRequestErrorHandler.java
+++ b/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 {
diff --git a/src/test/java/com/treasuredata/client/TDRequestErrorHandlerTest.java b/src/test/java/com/treasuredata/client/TDRequestErrorHandlerTest.java
index adc19f50..d467efaf 100644
--- a/src/test/java/com/treasuredata/client/TDRequestErrorHandlerTest.java
+++ b/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));
+    }
 }