Merge pull request #268 from Excaleo/configurable-retries

Make retryable exceptions configurable

Author: Mateusz Rzeszutek

signalfx-java/src/main/java/com/signalfx/connection/AbstractHttpReceiverConnection.java

@@ -1,6 +1,7 @@
 package com.signalfx.connection;
 
 import java.io.IOException;
+import java.util.List;
 import java.util.regex.Pattern;
 
 import org.apache.commons.io.IOUtils;
@@ -22,7 +23,8 @@
 import com.signalfx.endpoint.SignalFxReceiverEndpoint;
 import com.signalfx.metrics.SignalFxMetricsException;
 
-import static com.signalfx.connection.RetryHandler.DEFAULT_MAX_RETRIES;
+import static com.signalfx.connection.RetryDefaults.DEFAULT_MAX_RETRIES;
+import static com.signalfx.connection.RetryDefaults.DEFAULT_NON_RETRYABLE_EXCEPTIONS;
 
 public abstract class AbstractHttpReceiverConnection {
 
@@ -47,9 +49,14 @@ protected AbstractHttpReceiverConnection(SignalFxReceiverEndpoint endpoint, int
 
     protected AbstractHttpReceiverConnection(SignalFxReceiverEndpoint endpoint, int timeoutMs, int maxRetries,
                                              HttpClientConnectionManager httpClientConnectionManager) {
+        this(endpoint, timeoutMs, DEFAULT_MAX_RETRIES, httpClientConnectionManager, DEFAULT_NON_RETRYABLE_EXCEPTIONS);
+    }
+
+    protected AbstractHttpReceiverConnection(SignalFxReceiverEndpoint endpoint, int timeoutMs, int maxRetries,
+                                             HttpClientConnectionManager httpClientConnectionManager, List<Class<? extends IOException>> nonRetryableExceptions) {
         this.client = HttpClientBuilder.create()
                 .setConnectionManager(httpClientConnectionManager)
-                .setRetryHandler(new RetryHandler(maxRetries))
+                .setRetryHandler(new RetryHandler(maxRetries, nonRetryableExceptions))
                 .setServiceUnavailableRetryStrategy(new RetryStrategy(maxRetries))
                 .build();
         this.host = new HttpHost(endpoint.getHostname(), endpoint.getPort(), endpoint.getScheme());

signalfx-java/src/main/java/com/signalfx/connection/RetryDefaults.java

@@ -0,0 +1,20 @@
+package com.signalfx.connection;
+
+import java.io.IOException;
+import java.io.InterruptedIOException;
+import java.net.ConnectException;
+import java.net.UnknownHostException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+public final class RetryDefaults {
+    private RetryDefaults() {
+    }
+
+    public static final int DEFAULT_MAX_RETRIES = 3;
+    public static final List<Class<? extends IOException>> DEFAULT_NON_RETRYABLE_EXCEPTIONS = Collections.unmodifiableList(Arrays.asList(
+            InterruptedIOException.class,
+            UnknownHostException.class,
+            ConnectException.class));
+}

signalfx-java/src/main/java/com/signalfx/connection/RetryHandler.java

@@ -1,32 +1,30 @@
 package com.signalfx.connection;
 
-import java.io.InterruptedIOException;
-import java.net.ConnectException;
-import java.net.UnknownHostException;
-import java.util.Arrays;
+import java.io.IOException;
+import java.util.List;
 
 import org.apache.http.impl.client.DefaultHttpRequestRetryHandler;
 
+import static com.signalfx.connection.RetryDefaults.DEFAULT_MAX_RETRIES;
+import static com.signalfx.connection.RetryDefaults.DEFAULT_NON_RETRYABLE_EXCEPTIONS;
+
 /**
  * Compared to the {@link DefaultHttpRequestRetryHandler} we allow retry on {@link
  * javax.net.ssl.SSLException}, because it gets thrown when we try to send data points over a
  * connection that our server has already closed. It is still unknown how exactly our server closes
  * "stale" connections in such a way that http client is unable to detect this.
  */
 class RetryHandler extends DefaultHttpRequestRetryHandler {
-  public static final Integer DEFAULT_MAX_RETRIES = 3;
 
   public RetryHandler(final int maxRetries) {
-    super(maxRetries, true, Arrays.asList(
-        InterruptedIOException.class,
-        UnknownHostException.class,
-        ConnectException.class));
+    this(maxRetries, DEFAULT_NON_RETRYABLE_EXCEPTIONS);
   }
 
   public RetryHandler() {
-    super(DEFAULT_MAX_RETRIES, true, Arrays.asList(
-            InterruptedIOException.class,
-            UnknownHostException.class,
-            ConnectException.class));
+    this(DEFAULT_MAX_RETRIES, DEFAULT_NON_RETRYABLE_EXCEPTIONS);
+  }
+
+  public RetryHandler(final int maxRetries, List<Class<? extends IOException>> clazzes) {
+    super(maxRetries, true, clazzes);
   }
 }

signalfx-java/src/main/java/com/signalfx/metrics/connection/AbstractHttpDataPointProtobufReceiverConnection.java

@@ -1,28 +1,29 @@
 package com.signalfx.metrics.connection;
 
-import java.io.IOException;
-import java.nio.charset.StandardCharsets;
-import java.util.List;
-import java.util.Map;
-
 import com.google.common.collect.Lists;
+import com.signalfx.common.proto.ProtocolBufferStreamingInputStream;
+import com.signalfx.connection.AbstractHttpReceiverConnection;
+import com.signalfx.endpoint.SignalFxReceiverEndpoint;
+import com.signalfx.metrics.SignalFxMetricsException;
+import com.signalfx.metrics.protobuf.SignalFxProtocolBuffers;
 import org.apache.http.HttpEntity;
-import org.apache.http.HttpStatus;
 import org.apache.http.NameValuePair;
 import org.apache.http.client.methods.CloseableHttpResponse;
 import org.apache.http.client.utils.URLEncodedUtils;
 import org.apache.http.conn.HttpClientConnectionManager;
 import org.apache.http.entity.ContentType;
 import org.apache.http.entity.InputStreamEntity;
 import org.apache.http.message.BasicNameValuePair;
-
-import com.signalfx.common.proto.ProtocolBufferStreamingInputStream;
-import com.signalfx.connection.AbstractHttpReceiverConnection;
-import com.signalfx.endpoint.SignalFxReceiverEndpoint;
-import com.signalfx.metrics.SignalFxMetricsException;
-import com.signalfx.metrics.protobuf.SignalFxProtocolBuffers;
 import org.apache.http.util.EntityUtils;
 
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Map;
+
+import static com.signalfx.connection.RetryDefaults.DEFAULT_MAX_RETRIES;
+import static com.signalfx.connection.RetryDefaults.DEFAULT_NON_RETRYABLE_EXCEPTIONS;
+
 public abstract class AbstractHttpDataPointProtobufReceiverConnection extends AbstractHttpReceiverConnection implements DataPointReceiver {
 
     protected static final ContentType PROTO_TYPE = ContentType.create("application/x-protobuf");
@@ -32,15 +33,22 @@ public abstract class AbstractHttpDataPointProtobufReceiverConnection extends Ab
     public AbstractHttpDataPointProtobufReceiverConnection(SignalFxReceiverEndpoint endpoint,
                                                            int timeoutMs,
                                                            HttpClientConnectionManager httpClientConnectionManager) {
-        super(endpoint, timeoutMs, httpClientConnectionManager);
-        this.compress = !Boolean.getBoolean(DISABLE_COMPRESSION_PROPERTY);
+        this(endpoint, timeoutMs, DEFAULT_MAX_RETRIES, httpClientConnectionManager, DEFAULT_NON_RETRYABLE_EXCEPTIONS);
     }
 
     public AbstractHttpDataPointProtobufReceiverConnection(SignalFxReceiverEndpoint endpoint,
                                                            int timeoutMs,
                                                            int maxRetries,
                                                            HttpClientConnectionManager httpClientConnectionManager) {
-        super(endpoint, timeoutMs, maxRetries, httpClientConnectionManager);
+        this(endpoint, timeoutMs, maxRetries, httpClientConnectionManager, DEFAULT_NON_RETRYABLE_EXCEPTIONS);
+    }
+
+    public AbstractHttpDataPointProtobufReceiverConnection(SignalFxReceiverEndpoint endpoint,
+                                                           int timeoutMs,
+                                                           int maxRetries,
+                                                           HttpClientConnectionManager httpClientConnectionManager,
+                                                           List<Class<? extends IOException>> nonRetryableExceptions) {
+        super(endpoint, timeoutMs, maxRetries, httpClientConnectionManager, nonRetryableExceptions);
         this.compress = !Boolean.getBoolean(DISABLE_COMPRESSION_PROPERTY);
     }
 
@@ -84,7 +92,7 @@ protected abstract HttpEntity getEntityForVersion(
             List<SignalFxProtocolBuffers.DataPoint> dataPoints);
 
     @Override
-    public void backfillDataPoints(String auth, String metric, String metricType, String orgId, Map<String,String> dimensions,
+    public void backfillDataPoints(String auth, String metric, String metricType, String orgId, Map<String, String> dimensions,
                                    List<SignalFxProtocolBuffers.PointValue> datumPoints)
             throws SignalFxMetricsException {
         if (datumPoints.isEmpty()) {
@@ -97,7 +105,7 @@ public void backfillDataPoints(String auth, String metric, String metricType, St
         params.add(new BasicNameValuePair("metric", metric));
 
         // Each dimension is added as a param in the form of "sfxdim_DIMNAME"
-        for (Map.Entry<String,String> entry : dimensions.entrySet()) {
+        for (Map.Entry<String, String> entry : dimensions.entrySet()) {
             params.add(new BasicNameValuePair("sfxdim_" + entry.getKey(), entry.getValue()));
         }
 

signalfx-java/src/main/java/com/signalfx/metrics/connection/HttpDataPointProtobufReceiverConnection.java

@@ -37,6 +37,12 @@ public HttpDataPointProtobufReceiverConnection(
         super(endpoint, timeoutMs, maxRetries, httpClientConnectionManager);
     }
 
+    public HttpDataPointProtobufReceiverConnection(
+            SignalFxReceiverEndpoint endpoint, int timeoutMs, int maxRetries,
+            HttpClientConnectionManager httpClientConnectionManager, List<Class<? extends IOException>> nonRetryableExceptions) {
+        super(endpoint, timeoutMs, maxRetries, httpClientConnectionManager, nonRetryableExceptions);
+    }
+
     @Override
     protected HttpEntity getEntityForVersion(List<SignalFxProtocolBuffers.DataPoint> dataPoints) {
         return new InputStreamEntity(

signalfx-java/src/main/java/com/signalfx/metrics/connection/HttpDataPointProtobufReceiverConnectionV2.java

@@ -1,5 +1,6 @@
 package com.signalfx.metrics.connection;
 
+import java.io.IOException;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -26,6 +27,12 @@ public HttpDataPointProtobufReceiverConnectionV2(
         super(endpoint, timeoutMs, maxRetries, httpClientConnectionManager);
     }
 
+    public HttpDataPointProtobufReceiverConnectionV2(
+            SignalFxReceiverEndpoint endpoint, int timeoutMs, int maxRetries,
+            HttpClientConnectionManager httpClientConnectionManager, List<Class<? extends IOException>> nonRetryableExceptions) {
+        super(endpoint, timeoutMs, maxRetries, httpClientConnectionManager, nonRetryableExceptions);
+    }
+
     @Override
     protected String getEndpointForAddDatapoints() {
         return "/v2/datapoint";

signalfx-java/src/main/java/com/signalfx/metrics/connection/HttpDataPointProtobufReceiverFactory.java

@@ -6,17 +6,25 @@
 import com.signalfx.endpoint.SignalFxReceiverEndpoint;
 import com.signalfx.metrics.SignalFxMetricsException;
 
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import static com.signalfx.connection.RetryDefaults.DEFAULT_MAX_RETRIES;
+import static com.signalfx.connection.RetryDefaults.DEFAULT_NON_RETRYABLE_EXCEPTIONS;
+
 public class HttpDataPointProtobufReceiverFactory implements DataPointReceiverFactory {
     public static final int DEFAULT_TIMEOUT_MS = 2000;
     public static final int DEFAULT_VERSION = 2;
-    public static final int DEFAULT_MAX_RETRIES = 3;
 
     private final SignalFxReceiverEndpoint endpoint;
     private HttpClientConnectionManager httpClientConnectionManager;
     private HttpClientConnectionManager explicitHttpClientConnectionManager;
     private int timeoutMs = DEFAULT_TIMEOUT_MS;
     private int version = DEFAULT_VERSION;
     private int maxRetries = DEFAULT_MAX_RETRIES;
+    private List<Class<? extends IOException>> nonRetryableExceptions = DEFAULT_NON_RETRYABLE_EXCEPTIONS;
 
     public HttpDataPointProtobufReceiverFactory(SignalFxReceiverEndpoint endpoint) {
         this.endpoint = endpoint;
@@ -42,6 +50,11 @@ public HttpDataPointProtobufReceiverFactory setMaxRetries(int maxRetries) {
         return this;
     }
 
+    public HttpDataPointProtobufReceiverFactory setNonRetryableExceptions(List<Class<? extends IOException>> clazzes) {
+        this.nonRetryableExceptions = Collections.unmodifiableList(new ArrayList<>(clazzes));
+        return this;
+    }
+
     public void setHttpClientConnectionManager(
             HttpClientConnectionManager httpClientConnectionManager) {
         this.explicitHttpClientConnectionManager = httpClientConnectionManager;
@@ -55,13 +68,15 @@ public DataPointReceiver createDataPointReceiver() throws
                 endpoint,
                 this.timeoutMs,
                 this.maxRetries,
-                resolveHttpClientConnectionManager());
+                resolveHttpClientConnectionManager(),
+                this.nonRetryableExceptions);
         } else {
             return new HttpDataPointProtobufReceiverConnectionV2(
                 endpoint,
                 this.timeoutMs,
                 this.maxRetries,
-                resolveHttpClientConnectionManager());
+                resolveHttpClientConnectionManager(),
+                this.nonRetryableExceptions);
         }
 
     }

signalfx-java/src/test/java/com/signalfx/metrics/connection/HttpDataPointProtobufReceiverConnectionTest.java

@@ -14,8 +14,8 @@
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
+import java.util.concurrent.CountDownLatch;
 import java.util.zip.GZIPInputStream;
-import javax.servlet.ServletException;
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
 import org.apache.http.HttpStatus;
@@ -29,6 +29,10 @@
 import org.eclipse.jetty.server.handler.AbstractHandler;
 import org.junit.Test;
 
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
 public class HttpDataPointProtobufReceiverConnectionTest {
 
   public static final String AUTH_TOKEN = "AUTH_TOKEN";
@@ -95,6 +99,69 @@ public void shouldSendDataEvenIfServerClosedConnection() throws Exception {
     }
   }
 
+  @Test
+  public void shouldRetryOnSocketTimeout() throws Exception {
+    final CountDownLatch latch = new CountDownLatch(2);
+    final int clientTimeoutMs = 100;
+    final LatchedTimeoutHandler handler = new LatchedTimeoutHandler(latch, clientTimeoutMs);
+
+    Server server = new Server();
+    ServerConnector connector = new ServerConnector(server);
+    connector.setIdleTimeout(1000);
+    connector.setPort(0);
+    server.setConnectors(new Connector[]{connector});
+    server.setHandler(handler);
+    server.start();
+
+    try (AutoCloseable ignored = server::stop) {
+      URI uri = server.getURI();
+      DataPointReceiver dpr = new HttpDataPointProtobufReceiverFactory(
+              new SignalFxEndpoint(uri.getScheme(), uri.getHost(), uri.getPort()))
+              .setMaxRetries(1)
+              .setTimeoutMs(clientTimeoutMs)
+              .setNonRetryableExceptions(Collections.emptyList())
+              .createDataPointReceiver();
+      try {
+        dpr.addDataPoints(AUTH_TOKEN, Collections.singletonList(
+                SignalFxProtocolBuffers.DataPoint.newBuilder().setSource("source").build()));
+      } catch (Exception ignored2) {
+      }
+    }
+
+    assertTrue(latch.await(1000, MILLISECONDS));
+  }
+
+  @Test
+  public void shouldNotRetryOnDefaultNonRetryableExceptions() throws Exception {
+    final CountDownLatch latch = new CountDownLatch(1);
+    final int timeoutMs = 100;
+    final LatchedTimeoutHandler handler = new LatchedTimeoutHandler(latch, timeoutMs);
+
+    Server server = new Server();
+    ServerConnector connector = new ServerConnector(server);
+    connector.setIdleTimeout(1000);
+    connector.setPort(0);
+    server.setConnectors(new Connector[]{connector});
+    server.setHandler(handler);
+    server.start();
+
+    try (AutoCloseable ignored = server::stop) {
+      URI uri = server.getURI();
+      DataPointReceiver dpr = new HttpDataPointProtobufReceiverFactory(
+              new SignalFxEndpoint(uri.getScheme(), uri.getHost(), uri.getPort()))
+              .setTimeoutMs(timeoutMs)
+              .setMaxRetries(1)
+              .createDataPointReceiver();
+      try {
+        dpr.addDataPoints(AUTH_TOKEN, Collections.singletonList(
+                SignalFxProtocolBuffers.DataPoint.newBuilder().setSource("source").build()));
+      } catch (Exception ignored2) {
+      }
+    }
+
+    assertTrue(latch.await(1000, MILLISECONDS));
+  }
+
   @Test
   public void testBackfill() throws Exception {
     Server server = new Server(0);
@@ -193,6 +260,25 @@ public void handle(String target, Request baseRequest, HttpServletRequest reques
     }
   }
 
+  private static class LatchedTimeoutHandler extends AbstractHandler {
+    private final CountDownLatch latch;
+    private final int timeoutMs;
+
+    LatchedTimeoutHandler(CountDownLatch latch, int timeoutMs) {
+      this.latch = latch;
+      this.timeoutMs = timeoutMs;
+    }
+
+    @Override
+    public void handle(String s, Request request, HttpServletRequest httpServletRequest, HttpServletResponse httpServletResponse) throws IOException {
+      latch.countDown();
+      try {
+        Thread.sleep(timeoutMs);
+      } catch (Exception ignored) {
+      }
+    }
+  }
+
   private static void error(String message, HttpServletResponse response, Request baseRequest)
       throws IOException {
     response.setStatus(HttpStatus.SC_BAD_REQUEST);