Merge pull request #313 from DataDog/vickenty/fup

AGTMETRICS-489 Refactor forwarder to handle multiple URLs

Author: vickenty

dogstatsd-http-forwarder/src/main/java/com/datadoghq/dogstatsd/http/forwarder/BoundedQueue.java

@@ -54,7 +54,7 @@ public int compareTo(Key o) {
     final long maxTries;
     final WhenFull whenFull;
 
-    final TreeMap<Key, byte[]> items = new TreeMap<>();
+    final TreeMap<Key, Payload> items = new TreeMap<>();
 
     final Telemetry telemetry;
     final LongSupplier nanos;
@@ -88,14 +88,14 @@ private void recordDrop(long bytes) {
         droppedBytes += bytes;
     }
 
-    void add(byte[] item) throws InterruptedException {
+    void add(Payload item) throws InterruptedException {
         put(null, item, whenFull);
     }
 
-    void requeue(Map.Entry<Key, byte[]> item) throws InterruptedException {
+    void requeue(Map.Entry<Key, Payload> item) throws InterruptedException {
         Key nextKey = item.getKey().next();
         if (nextKey.tries > maxTries) {
-            telemetry.onDrop(1, item.getValue().length);
+            telemetry.onDrop(1, item.getValue().bytes.length);
             return;
         }
         put(nextKey, item.getValue(), WhenFull.DROP);
@@ -107,15 +107,15 @@ private Key newKey() {
         return new Key(0, clock, nanos.getAsLong());
     }
 
-    private void put(Key key, byte[] item, WhenFull whenFull) throws InterruptedException {
+    private void put(Key key, Payload item, WhenFull whenFull) throws InterruptedException {
         lock.lock();
         try {
             if (key == null) {
                 key = newKey();
             }
-            ensureSpace(item.length, whenFull);
+            ensureSpace(item.bytes.length, whenFull);
             items.put(key, item);
-            bytes += item.length;
+            bytes += item.bytes.length;
             notEmpty.signal();
         } finally {
             long droppedPayloads = this.droppedPayloads;
@@ -135,9 +135,9 @@ private void ensureSpace(int length, WhenFull whenFull) throws InterruptedExcept
         while (bytes + length > maxBytes) {
             switch (whenFull) {
                 case DROP:
-                    Map.Entry<Key, byte[]> last = items.pollLastEntry();
-                    bytes -= last.getValue().length;
-                    recordDrop(last.getValue().length);
+                    Map.Entry<Key, Payload> last = items.pollLastEntry();
+                    recordDrop(last.getValue().bytes.length);
+                    bytes -= last.getValue().bytes.length;
                     break;
                 case BLOCK:
                     notFull.await();
@@ -146,14 +146,14 @@ private void ensureSpace(int length, WhenFull whenFull) throws InterruptedExcept
         }
     }
 
-    Map.Entry<Key, byte[]> next() throws InterruptedException {
+    Map.Entry<Key, Payload> next() throws InterruptedException {
         lock.lock();
         try {
             while (items.size() == 0) {
                 notEmpty.await();
             }
-            Map.Entry<Key, byte[]> item = items.pollFirstEntry();
-            bytes -= item.getValue().length;
+            Map.Entry<Key, Payload> item = items.pollFirstEntry();
+            bytes -= item.getValue().bytes.length;
             notFull.signalAll();
             return item;
         } finally {

dogstatsd-http-forwarder/src/main/java/com/datadoghq/dogstatsd/http/forwarder/Forwarder.java

@@ -24,14 +24,13 @@
 /**
  * An HTTP forwarder that delivers DogStatsD HTTP payloads to a remote endpoint.
  *
- * <p>Payloads are enqueued via {@link #send(byte[])} and delivered asynchronously by a background
- * thread. Failed requests are retried with exponential back-off up to {@code maxTries} attempts
- * before being discarded.
+ * <p>Payloads are enqueued via {@link #send(URI, byte[])} and delivered asynchronously by a
+ * background thread. Failed requests are retried with exponential back-off up to {@code maxTries}
+ * attempts before being discarded.
  */
 public class Forwarder extends Thread {
     static final Logger logger = Logger.getLogger(Forwarder.class.getName());
     final BoundedQueue queue;
-    final URI url;
     final HttpClient client;
     final Duration requestTimeout;
     final Random rng = new Random();
@@ -42,9 +41,8 @@ public class Forwarder extends Thread {
     final Telemetry telemetry;
 
     /**
-     * Creates a new forwarder targeting the given URL.
+     * Creates a new forwarder.
      *
-     * @param url the remote HTTP endpoint to POST payloads to
      * @param maxRequestsBytes maximum total size of buffered payloads, in bytes
      * @param maxTries maximum number of delivery attempts per payload
      * @param whenFull action to take when the queue is at capacity
@@ -53,13 +51,11 @@ public class Forwarder extends Thread {
      *     {@code null} disables the request timeout
      */
     public Forwarder(
-            URI url,
             long maxRequestsBytes,
             long maxTries,
             WhenFull whenFull,
             Duration connectTimeout,
             Duration requestTimeout) {
-        this.url = url;
         this.telemetry = new Telemetry();
         this.queue = new BoundedQueue(maxRequestsBytes, maxTries, whenFull, this.telemetry);
         this.requestTimeout = requestTimeout;
@@ -91,26 +87,30 @@ public void run() {
     }
 
     /**
-     * Enqueues a payload for delivery to the remote endpoint.
+     * Enqueues a payload for delivery to the given endpoint.
      *
      * <p>If the queue is full, behaviour is determined by the {@link WhenFull} policy supplied at
      * construction time.
      *
+     * @param url the remote HTTP endpoint to POST the payload to
      * @param payload the raw bytes to deliver
      * @throws InterruptedException if the calling thread is interrupted while waiting for space
      *     ({@link WhenFull#BLOCK} mode only)
      */
-    public void send(byte[] payload) throws InterruptedException {
-        queue.add(payload);
+    public void send(URI url, byte[] payload) throws InterruptedException {
+        queue.add(new Payload(url, payload));
         telemetry.onEnqueue(payload.length);
     }
 
-    void runOnce(Map.Entry<BoundedQueue.Key, byte[]> item) throws InterruptedException {
-        byte[] payload = item.getValue();
-        logger.log(Level.INFO, "sending {0} bytes", payload.length);
+    void runOnce(Map.Entry<BoundedQueue.Key, Payload> item) throws InterruptedException {
+        Payload payload = item.getValue();
+        logger.log(
+                Level.INFO,
+                "sending {0} bytes to {1}",
+                new Object[] {payload.bytes.length, payload.url});
 
         HttpRequest.Builder builder =
-                HttpRequest.newBuilder(url).POST(BodyPublishers.ofByteArray(payload));
+                HttpRequest.newBuilder(payload.url).POST(BodyPublishers.ofByteArray(payload.bytes));
         if (requestTimeout != null) {
             builder.timeout(requestTimeout);
         }
@@ -138,9 +138,9 @@ void runOnce(Map.Entry<BoundedQueue.Key, byte[]> item) throws InterruptedExcepti
         backoff();
     }
 
-    void handleResponse(int code, Map.Entry<BoundedQueue.Key, byte[]> item)
+    void handleResponse(int code, Map.Entry<BoundedQueue.Key, Payload> item)
             throws InterruptedException {
-        int len = item.getValue().length;
+        int len = item.getValue().bytes.length;
         switch (code) {
             case 400:
                 telemetry.onResponse(code, len, false);
@@ -158,9 +158,9 @@ void handleResponse(int code, Map.Entry<BoundedQueue.Key, byte[]> item)
         }
     }
 
-    void handleTransportError(Map.Entry<BoundedQueue.Key, byte[]> item)
+    void handleTransportError(Map.Entry<BoundedQueue.Key, Payload> item)
             throws InterruptedException {
-        telemetry.onTransportError(item.getValue().length);
+        telemetry.onTransportError(item.getValue().bytes.length);
         increaseBackoff();
         queue.requeue(item);
     }

dogstatsd-http-forwarder/src/main/java/com/datadoghq/dogstatsd/http/forwarder/Payload.java

@@ -0,0 +1,20 @@
+/* Unless explicitly stated otherwise all files in this repository are
+ * licensed under the Apache 2.0 License.
+ *
+ * This product includes software developed at Datadog
+ *  (https://www.datadoghq.com/) Copyright 2026 Datadog, Inc.
+ */
+
+package com.datadoghq.dogstatsd.http.forwarder;
+
+import java.net.URI;
+
+class Payload {
+    final URI url;
+    final byte[] bytes;
+
+    Payload(URI url, byte[] bytes) {
+        this.url = url;
+        this.bytes = bytes;
+    }
+}