fix HttpPostSink initialization race condition

Author: BrandonArp

src/main/java/com/arpnetworking/tsdcore/sinks/HttpPostSink.java

@@ -45,6 +45,7 @@
 import org.asynchttpclient.RequestBuilder;
 import org.asynchttpclient.uri.Uri;
 
+import java.io.Serial;
 import java.net.URI;
 import java.time.Duration;
 import java.util.Collection;
@@ -206,6 +207,12 @@ protected HttpPostSink(final Builder<?, ?> builder) {
         _uri = builder._uri;
         _aysncHttpClientUri = Uri.create(_uri.toString());
 
+        _maximumAttempts = builder._maximumAttempts;
+        _baseBackoff = builder._baseBackoff;
+        _maximumDelay = builder._maximumDelay;
+        _acceptedStatusCodes = builder._acceptedStatusCodes;
+        _retryableStatusCodes = builder._retryableStatusCodes;
+
         _sinkActor = builder._actorSystem.actorOf(
                 HttpSinkActor.props(
                         CLIENT,
@@ -214,12 +221,6 @@ protected HttpPostSink(final Builder<?, ?> builder) {
                         builder._maximumQueueSize,
                         builder._spreadPeriod,
                         builder._periodicMetrics));
-        
-        _maximumAttempts = builder._maximumAttempts;
-        _baseBackoff = builder._baseBackoff;
-        _maximumDelay = builder._maximumDelay;
-        _acceptedStatusCodes = builder._acceptedStatusCodes;
-        _retryableStatusCodes = builder._retryableStatusCodes;
     }
 
     private final URI _uri;
@@ -441,6 +442,7 @@ protected Builder(final Function<B, S> targetConstructor) {
 
         private static final class CheckPeriod implements CheckWithCheck.SimpleCheck {
 
+            @Serial
             private static final long serialVersionUID = -6924010227680984149L;
 
             @Override

src/main/java/com/arpnetworking/tsdcore/sinks/HttpSinkActor.java

@@ -430,18 +430,30 @@ private void fireRequest(final RequestEntry request, final int attempt) {
         final long requestStartTime = System.currentTimeMillis();
         final CompletionStage<Object> responsePromise = promise
                 .handle((result, err) -> {
-                    _periodicMetrics.recordTimer(
-                            _requestLatencyName,
-                            System.currentTimeMillis() - requestStartTime,
-                            Optional.of(TimeUnit.MILLISECONDS));
-                    if (err == null) {
-                        if (_acceptedStatusCodes.contains(result.getStatusCode())) {
-                            return new PostSuccess(attempt, request, result);
+                    try {
+                        _periodicMetrics.recordTimer(
+                                _requestLatencyName,
+                                System.currentTimeMillis() - requestStartTime,
+                                Optional.of(TimeUnit.MILLISECONDS));
+                        if (err == null) {
+                            if (_acceptedStatusCodes.contains(result.getStatusCode())) {
+                                return new PostSuccess(attempt, request, result);
+                            } else {
+                                return new PostRejected(attempt, request, result);
+                            }
                         } else {
-                            return new PostRejected(attempt, request, result);
+                            return new PostFailure(attempt, request, err);
                         }
-                    } else {
-                        return new PostFailure(attempt, request, err);
+                        // CHECKSTYLE.OFF: IllegalCatch - We need to catch everything here
+                    } catch (final Exception e) {
+                        // CHECKSTYLE.ON: IllegalCatch
+                        POST_ERROR_LOGGER.error()
+                                .setMessage("Error while handling POST response")
+                                .setThrowable(e)
+                                .addData("sink", _sink)
+                                .addContext("actor", self())
+                                .log();
+                        return new PostFailure(attempt, request, e);
                     }
                 });
         Patterns.pipe(responsePromise, context().dispatcher()).to(self());