Add dogstatsd-http-forwarder

.circleci/config.yml

@@ -14,41 +14,54 @@ commands:
           name: Grab java version and dump to file
           command: java -version > << parameters.filename >>
 
-default_steps: &default_steps
+default_steps_7-11: &default_steps_7-11
   steps:
     - checkout
+    - run: mvn clean install
+    - run: cd dogstatsd-http-serializer && mvn clean install
 
-    - run: |
-        mvn clean install
+default_steps_11-21: &default_steps_11-21
+  steps:
+    - checkout
+    - run: mvn clean install
     - run: cd dogstatsd-http-serializer && mvn clean install
+    - run: cd dogstatsd-http-forwarder && mvn clean install
+
+default_steps_21-24: &default_steps_21-24
+  # Java 7 source code is not compatible with OpenJDK 21 and later.
+  steps:
+    - checkout
+    - run: cd dogstatsd-http-forwarder && mvn clean install
 
 jobs:
   openjdk7:
     docker:
       - image: jfullaondo/openjdk:7
-    <<: *default_steps
+    <<: *default_steps_7-11
   openjdk8:
     docker: &jdk8
       - image: cimg/openjdk:8.0
-    <<: *default_steps
+    <<: *default_steps_7-11
   openjdk11:
     docker:
       - image: cimg/openjdk:11.0
-    <<: *default_steps
+    <<: *default_steps_11-21
   openjdk13:
     docker:
       - image: cimg/openjdk:13.0
-    <<: *default_steps
+    <<: *default_steps_11-21
   openjdk17:
     docker:
       - image: cimg/openjdk:17.0
-    <<: *default_steps
-
-## Fails with "Source option 7 is no longer supported. Use 8 or later."
-#  openjdk21:
-#    docker:
-#      - image: cimg/openjdk:21.0
-#    <<: *default_steps
+    <<: *default_steps_11-21
+  openjdk21:
+    docker:
+      - image: cimg/openjdk:21.0
+    <<: *default_steps_21-24
+  openjdk24:
+    docker:
+      - image: cimg/openjdk:24.0
+    <<: *default_steps_21-24
 
   windows-openjdk12:
     executor:
@@ -85,7 +98,8 @@ workflows:
       - openjdk11
       - openjdk13
       - openjdk17
-#      - openjdk21
+      - openjdk21
+      - openjdk24
       - windows-openjdk12
       - openjdk8-jnr-exclude
       - openjdk8-jnr-latest

dogstatsd-http-forwarder/.gitignore

@@ -0,0 +1 @@
+/target/

dogstatsd-http-forwarder/pom.xml

@@ -0,0 +1,100 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <groupId>com.datadoghq</groupId>
+    <artifactId>dogstatsd-http-forwarder</artifactId>
+    <packaging>jar</packaging>
+    <name>dogstatsd-http-forwarder</name>
+    <version>1.0.0-SNAPSHOT</version>
+    <description>HTTP forwarder for DogStatsD metrics.</description>
+    <url>https://github.com/DataDog/java-dogstatsd-client</url>
+
+    <properties>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+    </properties>
+
+    <licenses>
+        <license>
+            <name>The MIT License (MIT)</name>
+            <url>http://opensource.org/licenses/MIT</url>
+            <distribution>repo</distribution>
+        </license>
+    </licenses>
+
+    <scm>
+        <url>https://github.com/DataDog/java-dogstatsd-client</url>
+        <connection>scm:git:git@github.com:DataDog/java-dogstatsd-client.git</connection>
+        <developerConnection>scm:git:git@github.com:Datadog/java-dogstatsd-client.git</developerConnection>
+    </scm>
+
+    <developers>
+        <developer>
+            <id>datadog</id>
+            <name>Datadog developers</name>
+            <email>dev@datadoghq.com</email>
+        </developer>
+    </developers>
+
+    <dependencies>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>4.13.1</version>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+    <profiles>
+        <profile>
+            <id>spotless</id>
+            <activation>
+                <jdk>[17.0,)</jdk>
+            </activation>
+            <build>
+                <plugins>
+                    <plugin>
+                        <groupId>com.diffplug.spotless</groupId>
+                        <artifactId>spotless-maven-plugin</artifactId>
+                        <version>2.45.0</version>
+                        <configuration>
+                            <java>
+                                <googleJavaFormat>
+                                    <version>1.28.0</version>
+                                    <style>AOSP</style>
+                                </googleJavaFormat>
+                            </java>
+                        </configuration>
+                        <executions>
+                            <execution>
+                                <goals>
+                                    <goal>check</goal>
+                                </goals>
+                            </execution>
+                        </executions>
+                    </plugin>
+                </plugins>
+            </build>
+        </profile>
+    </profiles>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <version>3.8.1</version>
+                <configuration>
+                    <source>11</source>
+                    <target>11</target>
+                </configuration>
+            </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-surefire-plugin</artifactId>
+                <version>2.19</version>
+            </plugin>
+        </plugins>
+    </build>
+</project>

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

@@ -0,0 +1,136 @@
+/* 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.util.Map;
+import java.util.TreeMap;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+class BoundedQueue {
+    // Key represents a tuple of integers (tries, clock).
+    static class Key implements Comparable<Key> {
+        final long tries;
+        final long clock;
+
+        Key(long clock) {
+            this.tries = 0;
+            this.clock = clock;
+        }
+
+        private Key(long tries, long clock) {
+            this.tries = tries;
+            this.clock = clock;
+        }
+
+        Key next() {
+            return new Key(tries + 1, clock);
+        }
+
+        @Override
+        public int compareTo(Key o) {
+            // Keys are ordered such first we try items with fewer
+            // attempts, and then with a newer (larger) clock value.
+            if (tries == o.tries) {
+                return Long.compare(o.clock, clock);
+            }
+            return Long.compare(tries, o.tries);
+        }
+    }
+
+    long clock = Long.MIN_VALUE;
+    long bytes;
+    final long maxBytes;
+    final long maxTries;
+    final WhenFull whenFull;
+
+    final TreeMap<Key, byte[]> items = new TreeMap<>();
+
+    long droppedItems;
+    long droppedBytes;
+
+    Lock lock = new ReentrantLock();
+    Condition notEmpty = lock.newCondition();
+    Condition notFull = lock.newCondition();
+
+    BoundedQueue(long maxBytes, long maxTries, WhenFull whenFull) {
+        this.maxBytes = maxBytes;
+        this.maxTries = maxTries;
+        this.whenFull = whenFull;
+    }
+
+    void add(byte[] item) throws InterruptedException {
+        put(null, item, whenFull);
+    }
+
+    void requeue(Map.Entry<Key, byte[]> item) throws InterruptedException {
+        Key nextKey = item.getKey().next();
+        if (nextKey.tries > maxTries) {
+            droppedItems++;
+            droppedBytes += item.getValue().length;
+            return;
+        }
+        put(nextKey, item.getValue(), WhenFull.DROP);
+    }
+
+    // Must be called when lock is held.
+    private Key newKey() {
+        clock++;
+        return new Key(clock);
+    }
+
+    private void put(Key key, byte[] item, WhenFull whenFull) throws InterruptedException {
+        lock.lock();
+        try {
+            if (key == null) {
+                key = newKey();
+            }
+            ensureSpace(item.length, whenFull);
+            items.put(key, item);
+            bytes += item.length;
+            notEmpty.signal();
+        } finally {
+            lock.unlock();
+        }
+    }
+
+    private void ensureSpace(int length, WhenFull whenFull) throws InterruptedException {
+        if (length > maxBytes) {
+            throw new IllegalArgumentException("item length is larger than maxBytes");
+        }
+        while (bytes + length > maxBytes) {
+            switch (whenFull) {
+                case DROP:
+                    Map.Entry<Key, byte[]> last = items.pollLastEntry();
+                    droppedItems++;
+                    droppedBytes += last.getValue().length;
+                    bytes -= last.getValue().length;
+                    break;
+                case BLOCK:
+                    notFull.await();
+                    break;
+            }
+        }
+    }
+
+    Map.Entry<Key, byte[]> next() throws InterruptedException {
+        lock.lock();
+        try {
+            while (items.size() == 0) {
+                notEmpty.await();
+            }
+            Map.Entry<Key, byte[]> item = items.pollFirstEntry();
+            bytes -= item.getValue().length;
+            notFull.signalAll();
+            return item;
+        } finally {
+            lock.unlock();
+        }
+    }
+}