dougqh · Oct 28, 2026 · Oct 28, 2026
@@ -24,6 +24,14 @@ protected void onCapacityExceeded() {
     throw new BufferOverflowException();
   }
 
+  /**
+   * A buffer whose chunks start at {@code firstChunkSize} and double, for usually-small payloads.
+   */
+  public ChunkedStreamingBuffer(int maxBytes, int firstChunkSize, ChunkedBufferConsumer consumer) {
+    super(maxBytes, firstChunkSize);
+    this.consumer = consumer;
+  }
+
   /**
    * Hands the completed messages to the consumer, then resets, which also releases the chunks the
    * payload didn't need.

@@ -375,7 +375,7 @@ abstract class InstrumentationSpecification extends DDSpecification implements A
     DDAgentFeaturesDiscovery features = new MockFeaturesDiscovery(true)
 
     Sink sink = new Sink() {
-      void accept(int messageCount, ByteBuffer buffer) {}
+      void accept(int messageCount, List<ByteBuffer> chunks) {}
 
       void register(EventListener listener) {}
     }

@@ -176,7 +176,7 @@ private static String repeat(char c, int count) {
 
   private static final class NoopSink implements Sink {
     @Override
-    public void accept(int messageCount, ByteBuffer buffer) {}
+    public void accept(int messageCount, List<ByteBuffer> chunks) {}
 
     @Override
     public void register(EventListener listener) {}

@@ -177,7 +177,7 @@ private static String repeat(char c, int count) {
 
   private static final class NoopSink implements Sink {
     @Override
-    public void accept(int messageCount, ByteBuffer buffer) {}
+    public void accept(int messageCount, List<ByteBuffer> chunks) {}
 
     @Override
     public void register(EventListener listener) {}

@@ -68,7 +68,7 @@ static class NullSink implements Sink {
     public void register(EventListener listener) {}
 
     @Override
-    public void accept(int messageCount, ByteBuffer buffer) {}
+    public void accept(int messageCount, List<ByteBuffer> chunks) {}
   }
 
   static class FixedAgentFeaturesDiscovery extends DDAgentFeaturesDiscovery {

@@ -1,6 +1,7 @@
 package datadog.trace.common.metrics;
 
 import java.nio.ByteBuffer;
+import java.util.List;
 
 /** A {@link Sink} that discards everything. */
 public final class NoOpSink implements Sink {
@@ -10,7 +11,7 @@ public final class NoOpSink implements Sink {
   private NoOpSink() {}
 
   @Override
-  public void accept(int messageCount, ByteBuffer buffer) {}
+  public void accept(int messageCount, List<ByteBuffer> chunks) {}
 
   @Override
   public void register(EventListener listener) {}

@@ -67,13 +67,13 @@ public OkHttpSink(
   }
 
   @Override
-  public void accept(int messageCount, ByteBuffer buffer) {
+  public void accept(int messageCount, List<ByteBuffer> chunks) {
     // if the agent is healthy, then we can send on this thread,
-    // without copying the buffer, otherwise this needs to be async,
+    // without copying the chunks, otherwise this needs to be async,
     // so need to copy and buffer the request, and let it be executed
     // on the main task scheduler as a last resort
     if (!bufferingEnabled || lastRequestTime.get() < ASYNC_THRESHOLD_LATENCY) {
-      send(prepareRequest(metricsUrl, headers).post(makeRequestBody(buffer)).build());
+      send(prepareRequest(metricsUrl, headers).post(makeRequestBody(chunks)).build());
       AgentTaskScheduler.Scheduled<OkHttpSink> future = this.future;
       if (future != null && enqueuedRequests.isEmpty()) {
         // async mode has been started but request latency is normal,
@@ -87,29 +87,46 @@ public void accept(int messageCount, ByteBuffer buffer) {
             AgentTaskScheduler.get()
                 .scheduleAtFixedRate(new Sender(enqueuedRequests), this, 1, 1, SECONDS);
       }
-      sendAsync(messageCount, buffer);
+      sendAsync(messageCount, chunks);
     }
   }
 
-  private RequestBody makeRequestBody(ByteBuffer buffer) {
+  private RequestBody makeRequestBody(List<ByteBuffer> chunks) {
     if (compressionEnabled) {
-      return gzippedMsgpackRequestBodyOf(Collections.singletonList(buffer));
+      return gzippedMsgpackRequestBodyOf(chunks);
     } else {
-      return msgpackRequestBodyOf(Collections.singletonList(buffer));
+      return msgpackRequestBodyOf(chunks);
     }
   }
 
-  private void sendAsync(int messageCount, ByteBuffer buffer) {
+  private void sendAsync(int messageCount, List<ByteBuffer> chunks) {
     asyncRequestCounter.getAndIncrement();
+    // the chunks are reused for the next payload as soon as accept returns, so copy them
+    ByteBuffer copy = copyOf(chunks);
     if (!enqueuedRequests.offer(
-        prepareRequest(metricsUrl, headers).post(makeRequestBody(buffer.duplicate())).build())) {
+        prepareRequest(metricsUrl, headers)
+            .post(makeRequestBody(Collections.singletonList(copy)))
+            .build())) {
       log.debug(
           "dropping payload of {} and {}B because sending queue was full",
           messageCount,
-          buffer.limit());
+          copy.limit());
     }
   }
 
+  private static ByteBuffer copyOf(List<ByteBuffer> chunks) {
+    int size = 0;
+    for (ByteBuffer chunk : chunks) {
+      size += chunk.remaining();
+    }
+    ByteBuffer copy = ByteBuffer.allocate(size);
+    for (ByteBuffer chunk : chunks) {
+      copy.put(chunk.duplicate());
+    }
+    copy.flip();
+    return copy;
+  }
+
   public boolean isInDegradedMode() {
     return asyncTaskStarted.get();
   }

@@ -3,7 +3,7 @@
 import static datadog.trace.bootstrap.instrumentation.api.UTF8BytesString.EMPTY;
 import static java.nio.charset.StandardCharsets.ISO_8859_1;
 
-import datadog.communication.serialization.GrowableBuffer;
+import datadog.communication.serialization.ChunkedStreamingBuffer;
 import datadog.communication.serialization.WritableFormatter;
 import datadog.communication.serialization.msgpack.MsgPackWriter;
 import datadog.metrics.api.Histogram;
@@ -14,12 +14,20 @@
 import datadog.trace.api.git.GitInfo;
 import datadog.trace.api.git.GitInfoProvider;
 import datadog.trace.bootstrap.instrumentation.api.UTF8BytesString;
+import datadog.trace.util.ChunkedBuffer;
+import java.nio.BufferOverflowException;
 import java.nio.ByteBuffer;
 import java.util.List;
 import java.util.function.Function;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public final class SerializingMetricWriter implements MetricWriter {
 
+  private static final Logger log = LoggerFactory.getLogger(SerializingMetricWriter.class);
+
+  private static final int MAX_PAYLOAD_SIZE = 5 << 20;
+
   private static final byte[] SEQUENCE = "Sequence".getBytes(ISO_8859_1);
   private static final byte[] RUNTIME_ID = "RuntimeID".getBytes(ISO_8859_1);
   private static final byte[] HOSTNAME = "Hostname".getBytes(ISO_8859_1);
@@ -73,7 +81,7 @@ public final class SerializingMetricWriter implements MetricWriter {
    */
   private final boolean additionalTagsConfigured;
 
-  private final GrowableBuffer buffer;
+  private final ChunkedStreamingBuffer buffer;
   private final DDCache<GitInfo, UTF8BytesString> gitInfoCache =
       DDCaches.newFixedSizeWeakKeyCache(4);
   private long sequence = 0;
@@ -91,37 +99,69 @@ public SerializingMetricWriter(WellKnownTags wellKnownTags, Sink sink) {
 
   public SerializingMetricWriter(
       WellKnownTags wellKnownTags, Sink sink, boolean additionalTagsConfigured) {
-    this(wellKnownTags, sink, 512 * 1024, GitInfoProvider.INSTANCE, additionalTagsConfigured);
+    this(
+        wellKnownTags,
+        sink,
+        ChunkedBuffer.DEFAULT_FIRST_CHUNK_SIZE,
+        GitInfoProvider.INSTANCE,
+        additionalTagsConfigured);
   }
 
-  public SerializingMetricWriter(WellKnownTags wellKnownTags, Sink sink, int initialCapacity) {
-    this(wellKnownTags, sink, initialCapacity, GitInfoProvider.INSTANCE);
+  public SerializingMetricWriter(WellKnownTags wellKnownTags, Sink sink, int firstChunkSize) {
+    this(wellKnownTags, sink, firstChunkSize, GitInfoProvider.INSTANCE);
   }
 
   public SerializingMetricWriter(
       WellKnownTags wellKnownTags,
       Sink sink,
-      int initialCapacity,
+      int firstChunkSize,
       final GitInfoProvider gitInfoProvider) {
-    this(wellKnownTags, sink, initialCapacity, gitInfoProvider, false);
+    this(wellKnownTags, sink, firstChunkSize, gitInfoProvider, false);
   }
 
   public SerializingMetricWriter(
       WellKnownTags wellKnownTags,
       Sink sink,
-      int initialCapacity,
+      int firstChunkSize,
+      final GitInfoProvider gitInfoProvider,
+      boolean additionalTagsConfigured) {
+    this(
+        wellKnownTags,
+        sink,
+        firstChunkSize,
+        MAX_PAYLOAD_SIZE,
+        gitInfoProvider,
+        additionalTagsConfigured);
+  }
+
+  SerializingMetricWriter(
+      WellKnownTags wellKnownTags,
+      Sink sink,
+      int firstChunkSize,
+      int maxPayloadSize,
       final GitInfoProvider gitInfoProvider,
       boolean additionalTagsConfigured) {
     this.wellKnownTags = wellKnownTags;
-    this.buffer = new GrowableBuffer(initialCapacity);
+    this.buffer = new ChunkedStreamingBuffer(maxPayloadSize, firstChunkSize, sink);
     this.writer = new MsgPackWriter(buffer);
     this.sink = sink;
     this.gitInfoProvider = gitInfoProvider;
     this.additionalTagsConfigured = additionalTagsConfigured;
   }
 
+  // A payload spans startBucket, add and finishBucket. Only the first write past the cap throws;
+  // the buffer discards the rest, and mark() in finishBucket reports the payload as dropped.
+
   @Override
   public void startBucket(int metricCount, long start, long duration) {
+    try {
+      writeBucketStart(metricCount, start, duration);
+    } catch (BufferOverflowException ignored) {
+      // reported by finishBucket
+    }
+  }
+
+  private void writeBucketStart(int metricCount, long start, long duration) {
     final UTF8BytesString processTags = ProcessTags.getTagsForSerialization();
     final boolean writeProcessTags = processTags != null;
     // the gitinfo can be rebuild in time and initialized after this class is created. So that a
@@ -177,6 +217,14 @@ public void startBucket(int metricCount, long start, long duration) {
 
   @Override
   public void add(AggregateEntry entry) {
+    try {
+      writeEntry(entry);
+    } catch (BufferOverflowException ignored) {
+      // reported by finishBucket
+    }
+  }
+
+  private void writeEntry(AggregateEntry entry) {
     // Dynamic map size based on optional fields; AggregateEntry encapsulates the EMPTY-as-absent
     // sentinel via its hasFoo() predicates so the serializer doesn't depend on the storage choice.
     final boolean hasHttpMethod = entry.hasHttpMethod();
@@ -306,9 +354,12 @@ private byte[] emptyErrorHistogramBytes() {
 
   @Override
   public void finishBucket() {
-    buffer.mark();
-    sink.accept(buffer.messageCount(), buffer.slice());
-    buffer.reset();
+    if (buffer.mark()) {
+      buffer.flush();
+    } else {
+      log.debug("dropping a stats payload larger than {} bytes", buffer.capacity());
+      buffer.reset();
+    }
   }
 
   @Override

@@ -1,8 +1,15 @@
 package datadog.trace.common.metrics;
 
-import datadog.communication.serialization.ByteBufferConsumer;
+import datadog.communication.serialization.ChunkedBufferConsumer;
+import java.nio.ByteBuffer;
+import java.util.Collections;
 
-public interface Sink extends ByteBufferConsumer {
+public interface Sink extends ChunkedBufferConsumer {
+
+  /** Accepts a payload held in a single buffer. */
+  default void accept(int messageCount, ByteBuffer buffer) {
+    accept(messageCount, Collections.singletonList(buffer));
+  }
 
   void register(EventListener listener);
 }
@@ -16,6 +16,7 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Random;
 import java.util.UUID;
@@ -177,7 +178,7 @@ private static class ValidatingSink implements Sink {
     public void register(EventListener listener) {}
 
     @Override
-    public void accept(int messageCount, ByteBuffer buffer) {
+    public void accept(int messageCount, List<ByteBuffer> chunks) {
       latch.countDown();
     }
   }