From 02c52dfcc34a429903e4b2de4808be8387260024 Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Wed, 28 Oct 2026 09:06:22 -0400 Subject: [PATCH 1/2] Serialize client stats payloads into a chunked buffer SerializingMetricWriter held a 512 KiB GrowableBuffer for the tracer's lifetime: a G1 humongous object on heaps up to about 2 GB (512 KiB plus the array header is just over half a 1 MB region). - ChunkedBuffer can start with small chunks that double up to 256 KiB, so payloads of a few KB keep a few KB. - SerializingMetricWriter uses a ChunkedStreamingBuffer with a 5 MiB cap: nothing is allocated until the first payload. The first write past the cap throws and is caught; the buffer discards the rest, and finishBucket drops the payload with a debug log. - Sink takes the payload as chunks (a single-buffer default remains for the DSM writer until it moves too). OkHttpSink sends the chunks directly on the synchronous path and copies them on the async path. The async path claimed to copy but only duplicated the buffer, so a queued request shared memory the writer reused for the next payload. Co-Authored-By: Claude Opus 5.5 --- .../serialization/ChunkedStreamingBuffer.java | 8 +++ .../test/InstrumentationSpecification.groovy | 2 +- ...aProducerFailureDataStreamsForkedTest.java | 2 +- ...aProducerFailureDataStreamsForkedTest.java | 2 +- .../ClientStatsAggregatorBenchmark.java | 2 +- .../trace/common/metrics/NoOpSink.java | 3 +- .../trace/common/metrics/OkHttpSink.java | 37 +++++++--- .../metrics/SerializingMetricWriter.java | 71 +++++++++++++++---- .../datadog/trace/common/metrics/Sink.java | 11 ++- .../common/metrics/FootprintForkedTest.java | 3 +- .../trace/common/metrics/OkHttpSinkTest.java | 52 ++++++++++++-- .../metrics/SerializingMetricWriterTest.java | 68 +++++++++++++++++- .../datadog/trace/util/ChunkedBuffer.java | 47 ++++++++++-- .../datadog/trace/util/ChunkedBufferTest.java | 46 ++++++++++++ 14 files changed, 313 insertions(+), 41 deletions(-) diff --git a/communication/src/main/java/datadog/communication/serialization/ChunkedStreamingBuffer.java b/communication/src/main/java/datadog/communication/serialization/ChunkedStreamingBuffer.java index 6a82a77b49f..f95a7895936 100644 --- a/communication/src/main/java/datadog/communication/serialization/ChunkedStreamingBuffer.java +++ b/communication/src/main/java/datadog/communication/serialization/ChunkedStreamingBuffer.java @@ -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. diff --git a/dd-java-agent/instrumentation-testing/src/main/groovy/datadog/trace/agent/test/InstrumentationSpecification.groovy b/dd-java-agent/instrumentation-testing/src/main/groovy/datadog/trace/agent/test/InstrumentationSpecification.groovy index 38904cdd717..9e5b9246d23 100644 --- a/dd-java-agent/instrumentation-testing/src/main/groovy/datadog/trace/agent/test/InstrumentationSpecification.groovy +++ b/dd-java-agent/instrumentation-testing/src/main/groovy/datadog/trace/agent/test/InstrumentationSpecification.groovy @@ -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 chunks) {} void register(EventListener listener) {} } diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/latestDepTest/java/KafkaProducerFailureDataStreamsForkedTest.java b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/latestDepTest/java/KafkaProducerFailureDataStreamsForkedTest.java index 2ab9e0b7763..4091a17dd2f 100644 --- a/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/latestDepTest/java/KafkaProducerFailureDataStreamsForkedTest.java +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/latestDepTest/java/KafkaProducerFailureDataStreamsForkedTest.java @@ -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 chunks) {} @Override public void register(EventListener listener) {} diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/test/java/KafkaProducerFailureDataStreamsForkedTest.java b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/test/java/KafkaProducerFailureDataStreamsForkedTest.java index b56ed2328fa..867b65cf7a8 100644 --- a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/test/java/KafkaProducerFailureDataStreamsForkedTest.java +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/test/java/KafkaProducerFailureDataStreamsForkedTest.java @@ -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 chunks) {} @Override public void register(EventListener listener) {} diff --git a/dd-trace-core/src/jmh/java/datadog/trace/common/metrics/ClientStatsAggregatorBenchmark.java b/dd-trace-core/src/jmh/java/datadog/trace/common/metrics/ClientStatsAggregatorBenchmark.java index 08449d7a9ac..5c9420283e9 100644 --- a/dd-trace-core/src/jmh/java/datadog/trace/common/metrics/ClientStatsAggregatorBenchmark.java +++ b/dd-trace-core/src/jmh/java/datadog/trace/common/metrics/ClientStatsAggregatorBenchmark.java @@ -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 chunks) {} } static class FixedAgentFeaturesDiscovery extends DDAgentFeaturesDiscovery { diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/NoOpSink.java b/dd-trace-core/src/main/java/datadog/trace/common/metrics/NoOpSink.java index 75236791324..eaca6935f32 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/metrics/NoOpSink.java +++ b/dd-trace-core/src/main/java/datadog/trace/common/metrics/NoOpSink.java @@ -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 chunks) {} @Override public void register(EventListener listener) {} diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/OkHttpSink.java b/dd-trace-core/src/main/java/datadog/trace/common/metrics/OkHttpSink.java index 193a2fb0436..d1c93eefc0c 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/metrics/OkHttpSink.java +++ b/dd-trace-core/src/main/java/datadog/trace/common/metrics/OkHttpSink.java @@ -67,13 +67,13 @@ public OkHttpSink( } @Override - public void accept(int messageCount, ByteBuffer buffer) { + public void accept(int messageCount, List 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 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 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 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 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(); } diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/SerializingMetricWriter.java b/dd-trace-core/src/main/java/datadog/trace/common/metrics/SerializingMetricWriter.java index 3eacde8bd45..03c7e31960b 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/metrics/SerializingMetricWriter.java +++ b/dd-trace-core/src/main/java/datadog/trace/common/metrics/SerializingMetricWriter.java @@ -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,21 @@ import datadog.trace.api.git.GitInfo; import datadog.trace.api.git.GitInfoProvider; import datadog.trace.bootstrap.instrumentation.api.UTF8BytesString; +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); + + // payloads are usually a few KB: start small, and nothing is allocated until the first payload + private static final int FIRST_CHUNK_SIZE = 4 * 1024; + 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 +82,7 @@ public final class SerializingMetricWriter implements MetricWriter { */ private final boolean additionalTagsConfigured; - private final GrowableBuffer buffer; + private final ChunkedStreamingBuffer buffer; private final DDCache gitInfoCache = DDCaches.newFixedSizeWeakKeyCache(4); private long sequence = 0; @@ -91,37 +100,64 @@ 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, 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 +213,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 +350,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 diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/Sink.java b/dd-trace-core/src/main/java/datadog/trace/common/metrics/Sink.java index 7a812f341f4..f8261339926 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/metrics/Sink.java +++ b/dd-trace-core/src/main/java/datadog/trace/common/metrics/Sink.java @@ -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); } diff --git a/dd-trace-core/src/test/java/datadog/trace/common/metrics/FootprintForkedTest.java b/dd-trace-core/src/test/java/datadog/trace/common/metrics/FootprintForkedTest.java index ac54d7d700f..78dbed26f3b 100644 --- a/dd-trace-core/src/test/java/datadog/trace/common/metrics/FootprintForkedTest.java +++ b/dd-trace-core/src/test/java/datadog/trace/common/metrics/FootprintForkedTest.java @@ -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 chunks) { latch.countDown(); } } diff --git a/dd-trace-core/src/test/java/datadog/trace/common/metrics/OkHttpSinkTest.java b/dd-trace-core/src/test/java/datadog/trace/common/metrics/OkHttpSinkTest.java index c864baeaf8c..044ac8d78cb 100644 --- a/dd-trace-core/src/test/java/datadog/trace/common/metrics/OkHttpSinkTest.java +++ b/dd-trace-core/src/test/java/datadog/trace/common/metrics/OkHttpSinkTest.java @@ -3,6 +3,7 @@ import static datadog.communication.ddagent.DDAgentFeaturesDiscovery.V06_METRICS_ENDPOINT; import static datadog.trace.common.metrics.EventListener.EventType.OK; import static java.util.concurrent.TimeUnit.SECONDS; +import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -17,6 +18,7 @@ import java.io.IOException; import java.nio.ByteBuffer; +import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.concurrent.CopyOnWriteArrayList; @@ -55,7 +57,7 @@ void httpStatusCodeResponseCodeYieldsEventType( doAnswer(invocation -> respond(invocation.getArgument(0), responseCode)) .when(client) .newCall(any()); - sink.accept(0, ByteBuffer.allocate(0)); + sink.accept(0, Collections.singletonList(ByteBuffer.allocate(0))); verify(client, times(1)).newCall(any()); verify(listener).onEvent(eq(eventType), any()); @@ -97,8 +99,8 @@ void degradeToAsyncModeWhenAgentSlowToRespond() throws Exception { .newCall(any()); // one slow response followed by a request - sink.accept(1, ByteBuffer.allocate(0)); - sink.accept(1, ByteBuffer.allocate(0)); + sink.accept(1, Collections.singletonList(ByteBuffer.allocate(0))); + sink.accept(1, Collections.singletonList(ByteBuffer.allocate(0))); latch.await(10, SECONDS); // the second request degrades to async mode @@ -112,7 +114,7 @@ void degradeToAsyncModeWhenAgentSlowToRespond() throws Exception { assertTrue(sink.isInDegradedMode()); // the agent has recovered and has responded quickly once - sink.accept(1, ByteBuffer.allocate(0)); + sink.accept(1, Collections.singletonList(ByteBuffer.allocate(0))); // the request was sent synchronously verify(client, times(3)).newCall(any()); @@ -120,6 +122,48 @@ void degradeToAsyncModeWhenAgentSlowToRespond() throws Exception { assertFalse(sink.isInDegradedMode()); } + @Test + void asyncModeSendsACopyOfThePayload() throws Exception { + // the writer reuses its chunks for the next payload as soon as accept returns, so a request + // queued in async mode must not share them + String agentUrl = "http://localhost:8126"; + CountDownLatch latch = new CountDownLatch(2); + BlockingListener listener = new BlockingListener(latch); + OkHttpClient client = mock(OkHttpClient.class); + OkHttpSink sink = + new OkHttpSink(client, agentUrl, V06_METRICS_ENDPOINT, true, false, Collections.emptyMap()); + sink.register(listener); + AtomicInteger callCount = new AtomicInteger(0); + List sentBodies = new CopyOnWriteArrayList<>(); + doAnswer( + invocation -> { + Request request = invocation.getArgument(0); + okio.Buffer body = new okio.Buffer(); + request.body().writeTo(body); + sentBodies.add(body.readByteArray()); + if (callCount.incrementAndGet() == 1) { + // a slow agent switches the sink to async mode + Thread.sleep(1001); + } + return respond(request, 200); + }) + .when(client) + .newCall(any()); + + sink.accept(1, Collections.singletonList(ByteBuffer.wrap(new byte[] {1}))); + byte[] first = {2, 3}; + byte[] second = {4, 5}; + sink.accept(1, Arrays.asList(ByteBuffer.wrap(first), ByteBuffer.wrap(second))); + assertTrue(sink.isInDegradedMode()); + // the writer starts its next payload in the same chunks + Arrays.fill(first, (byte) 9); + Arrays.fill(second, (byte) 9); + latch.await(10, SECONDS); + + assertEquals(2, sentBodies.size()); + assertArrayEquals(new byte[] {2, 3, 4, 5}, sentBodies.get(1)); + } + private static Call respond(Request request, int code) throws IOException { if (code == 0) { return error(); diff --git a/dd-trace-core/src/test/java/datadog/trace/common/metrics/SerializingMetricWriterTest.java b/dd-trace-core/src/test/java/datadog/trace/common/metrics/SerializingMetricWriterTest.java index bc05393cea3..84b8c5f93a0 100644 --- a/dd-trace-core/src/test/java/datadog/trace/common/metrics/SerializingMetricWriterTest.java +++ b/dd-trace-core/src/test/java/datadog/trace/common/metrics/SerializingMetricWriterTest.java @@ -620,6 +620,57 @@ private static void serializeAndValidate(List content) { assertTrue(sink.validatedInput()); } + @Test + void payloadOverTheCapIsDroppedAndTheNextOneIsSent() { + long startTime = MILLISECONDS.toNanos(System.currentTimeMillis()); + long duration = SECONDS.toNanos(10); + WellKnownTags wellKnownTags = + new WellKnownTags("runtimeid", "hostname", "env", "service", "version", "language"); + AggregateEntry entry = + entry( + "resource", + "service", + "operation", + null, + "type", + 200, + false, + false, + "server", + Collections.emptyList(), + "GET", + "/api/users", + null, + 1); + List sent = new ArrayList<>(); + Sink sink = + new Sink() { + @Override + public void accept(int messageCount, List chunks) { + sent.add(messageCount); + } + + @Override + public void register(EventListener listener) {} + }; + SerializingMetricWriter writer = + new SerializingMetricWriter( + wellKnownTags, sink, 128, 4096, GitInfoProvider.INSTANCE, false); + + // a bucket well past the 4 KiB cap is dropped, without throwing + writer.startBucket(2000, startTime, duration); + for (int i = 0; i < 2000; ++i) { + writer.add(entry); + } + writer.finishBucket(); + assertTrue(sent.isEmpty()); + + writer.startBucket(1, startTime, duration); + writer.add(entry); + writer.finishBucket(); + assertEquals(Collections.singletonList(1), sent); + } + static final class ValidatingSink implements Sink { private final WellKnownTags wellKnownTags; @@ -666,16 +717,29 @@ static final class ValidatingSink implements Sink { public void register(EventListener listener) {} @Override - public void accept(int messageCount, ByteBuffer buffer) { + public void accept(int messageCount, List chunks) { // Sink.accept can't declare checked exceptions; surface any msgpack decode failure as an // assertion error so the test reports it. try { - validate(buffer); + validate(join(chunks)); } catch (IOException e) { throw new AssertionError("Failed to decode metric payload", e); } } + private static ByteBuffer join(List chunks) { + int size = 0; + for (ByteBuffer chunk : chunks) { + size += chunk.remaining(); + } + ByteBuffer joined = ByteBuffer.allocate(size); + for (ByteBuffer chunk : chunks) { + joined.put(chunk.duplicate()); + } + joined.flip(); + return joined; + } + private void validate(ByteBuffer buffer) throws IOException { MessageUnpacker unpacker = MessagePack.newDefaultUnpacker(buffer); int mapSize = unpacker.unpackMapHeader(); diff --git a/internal-api/src/main/java/datadog/trace/util/ChunkedBuffer.java b/internal-api/src/main/java/datadog/trace/util/ChunkedBuffer.java index c9e64e3d60e..475f790c660 100644 --- a/internal-api/src/main/java/datadog/trace/util/ChunkedBuffer.java +++ b/internal-api/src/main/java/datadog/trace/util/ChunkedBuffer.java @@ -13,8 +13,8 @@ * collector: G1 treats anything over half a region as humongous (1 MB regions on heaps up to about * 2 GB), Shenandoah anything over a region, and ZGC moves anything over 256 KiB off its small * pages. Humongous objects sit in old gen, are never compacted, and push the collector toward - * concurrent cycles. Each chunk here is an array of {@link #CHUNK_SIZE} bytes, a whole object of at - * most 256 KiB for any array header layout, which is regular-sized on all of them. + * concurrent cycles. Each chunk here is an array of at most {@link #CHUNK_SIZE} bytes, a whole + * object of at most 256 KiB for any array header layout, which is regular-sized on all of them. * *

Messages, marks and rewinding. Callers write a message, then call {@link #mark()} to * complete it. {@link #rewind()} drops everything written since the last mark, so a message that @@ -36,6 +36,10 @@ * so the buffer holds exactly as much as a single array of the same capacity would. The last chunk * is only as long as the remaining capacity, so a small buffer is a single small chunk. * + *

Small payloads stay small. Chunks can start smaller than {@link #CHUNK_SIZE} and double + * up to it (see {@link #ChunkedBuffer(int, int)}), so a buffer whose payloads are usually a few KB + * holds a few KB rather than a whole chunk. + * *

Retention. Chunks are allocated on first use. {@link #reset()} keeps the chunks that * held completed data plus half of any surplus kept from the previous reset, so after a burst the * extra chunks drain over a few resets instead of all at once. Payloads that vary in size don't @@ -57,7 +61,11 @@ public class ChunkedBuffer { /** Where writes go once the buffer has failed; reused from the start whenever it fills. */ private static final int SCRATCH_SIZE = 64; + /** The smallest first chunk: large enough that any primitive fits in a fresh chunk. */ + private static final int MIN_FIRST_CHUNK_SIZE = 64; + private final int maxBytes; + private final int firstChunkSize; private final int maxChunks; private final byte[][] chunks; private final ByteBuffer[] views; @@ -88,9 +96,18 @@ public class ChunkedBuffer { * @param maxBytes the most data the buffer may hold */ public ChunkedBuffer(int maxBytes) { + this(maxBytes, CHUNK_SIZE); + } + + /** + * @param maxBytes the most data the buffer may hold + * @param firstChunkSize the size of the first chunk; each later chunk doubles, up to {@link + * #CHUNK_SIZE} + */ + public ChunkedBuffer(int maxBytes, int firstChunkSize) { this.maxBytes = Math.max(0, maxBytes); - // each chunk may leave up to 7 bytes unused, so allow for one more chunk than whole chunks - this.maxChunks = this.maxBytes == 0 ? 0 : this.maxBytes / (CHUNK_SIZE - 7) + 1; + this.firstChunkSize = Math.max(MIN_FIRST_CHUNK_SIZE, Math.min(CHUNK_SIZE, firstChunkSize)); + this.maxChunks = chunksNeeded(); this.chunks = new byte[maxChunks][]; this.views = new ByteBuffer[maxChunks]; this.fills = new int[maxChunks]; @@ -306,7 +323,7 @@ private int advance(int width) { fail(); return 0; } - int length = Math.min(CHUNK_SIZE, remaining); + int length = Math.min(chunkSize(next), remaining); if (next == allocatedChunks) { ++allocatedChunks; allocate(next, length); @@ -320,6 +337,26 @@ private int advance(int width) { return 0; } + /** The full size of the chunk at {@code index}: doubling from the first chunk, up to the max. */ + private int chunkSize(int index) { + long size = (long) firstChunkSize << Math.min(index, 31); + return (int) Math.min(CHUNK_SIZE, size); + } + + /** Enough chunks to hold {@code maxBytes} even if each leaves up to 7 bytes unused. */ + private int chunksNeeded() { + if (maxBytes == 0) { + return 0; + } + long covered = 0; + int count = 0; + while (covered < maxBytes) { + covered += chunkSize(count) - 7; + ++count; + } + return count + 1; + } + private void allocate(int index, int length) { byte[] chunk = new byte[length]; chunks[index] = chunk; diff --git a/internal-api/src/test/java/datadog/trace/util/ChunkedBufferTest.java b/internal-api/src/test/java/datadog/trace/util/ChunkedBufferTest.java index 4ee50d1475b..dd4cf59b013 100644 --- a/internal-api/src/test/java/datadog/trace/util/ChunkedBufferTest.java +++ b/internal-api/src/test/java/datadog/trace/util/ChunkedBufferTest.java @@ -404,6 +404,52 @@ protected void onCapacityExceeded() { assertEquals(8, buffer.completedBytes()); } + @Test + void chunksStartSmallAndDoubleUpToChunkSize() { + ChunkedBuffer buffer = new ChunkedBuffer(4 * CHUNK_SIZE, 4096); + buffer.put(new byte[4096 + 8192 + 16384 + 32768 + 65536 + 131072 + CHUNK_SIZE + 1]); + assertTrue(buffer.mark()); + + List chunks = buffer.completedChunks(); + int[] expected = {4096, 8192, 16384, 32768, 65536, 131072, CHUNK_SIZE, 1}; + assertEquals(expected.length, chunks.size()); + for (int i = 0; i < expected.length; ++i) { + assertEquals(expected[i], chunks.get(i).remaining(), "chunk " + i); + } + } + + @Test + void smallPayloadKeepsOnlyASmallChunk() { + ChunkedBuffer buffer = new ChunkedBuffer(32 << 20, 4096); + buffer.put(new byte[100]); + assertTrue(buffer.mark()); + assertEquals(4096, buffer.completedChunks().get(0).array().length); + buffer.reset(); + assertEquals(1, buffer.allocatedChunks()); + } + + @Test + void capacityIsExactWithSmallFirstChunks() { + ChunkedBuffer buffer = new ChunkedBuffer(300_000, 4096); + // primitives leave unused tails in the small chunks + for (int i = 0; i < 300_000 / 8; ++i) { + buffer.putLong(i); + } + assertTrue(buffer.mark()); + assertEquals(300_000, buffer.completedBytes()); + + buffer.put((byte) 1); + assertFalse(buffer.mark()); + } + + @Test + void firstChunkSizeIsLargeEnoughForAnyPrimitive() { + ChunkedBuffer buffer = new ChunkedBuffer(1024, 1); + buffer.putLong(7L); + assertTrue(buffer.mark()); + assertEquals(7L, buffer.completedChunks().get(0).getLong()); + } + private static byte[] sequence(int length) { byte[] data = new byte[length]; for (int i = 0; i < length; ++i) { From dcc3b8d8db8ab65637d6d19d056ee0c1c8db304b Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Wed, 28 Oct 2026 09:06:32 -0400 Subject: [PATCH 2/2] Start chunked buffers with small chunks by default ChunkedBuffer(maxBytes) now starts at a 4 KiB chunk and doubles up to 256 KiB, so callers pass only a cap. A quiet buffer, including the trace writer on a quiet service, keeps a few KB instead of a 256 KiB chunk; a large payload reaches full-size chunks after about 250 KB, a few more slow-path steps per payload. The stats writer uses the shared default. Co-Authored-By: Claude Opus 5.5 --- .../metrics/SerializingMetricWriter.java | 10 ++- .../datadog/trace/util/ChunkedBuffer.java | 11 ++-- .../datadog/trace/util/ChunkedBufferTest.java | 61 +++++++++++-------- 3 files changed, 49 insertions(+), 33 deletions(-) diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/SerializingMetricWriter.java b/dd-trace-core/src/main/java/datadog/trace/common/metrics/SerializingMetricWriter.java index 03c7e31960b..db303233b30 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/metrics/SerializingMetricWriter.java +++ b/dd-trace-core/src/main/java/datadog/trace/common/metrics/SerializingMetricWriter.java @@ -14,6 +14,7 @@ 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; @@ -25,8 +26,6 @@ public final class SerializingMetricWriter implements MetricWriter { private static final Logger log = LoggerFactory.getLogger(SerializingMetricWriter.class); - // payloads are usually a few KB: start small, and nothing is allocated until the first payload - private static final int FIRST_CHUNK_SIZE = 4 * 1024; private static final int MAX_PAYLOAD_SIZE = 5 << 20; private static final byte[] SEQUENCE = "Sequence".getBytes(ISO_8859_1); @@ -100,7 +99,12 @@ public SerializingMetricWriter(WellKnownTags wellKnownTags, Sink sink) { public SerializingMetricWriter( WellKnownTags wellKnownTags, Sink sink, boolean additionalTagsConfigured) { - this(wellKnownTags, sink, FIRST_CHUNK_SIZE, GitInfoProvider.INSTANCE, additionalTagsConfigured); + this( + wellKnownTags, + sink, + ChunkedBuffer.DEFAULT_FIRST_CHUNK_SIZE, + GitInfoProvider.INSTANCE, + additionalTagsConfigured); } public SerializingMetricWriter(WellKnownTags wellKnownTags, Sink sink, int firstChunkSize) { diff --git a/internal-api/src/main/java/datadog/trace/util/ChunkedBuffer.java b/internal-api/src/main/java/datadog/trace/util/ChunkedBuffer.java index 475f790c660..dd5a1bde8d9 100644 --- a/internal-api/src/main/java/datadog/trace/util/ChunkedBuffer.java +++ b/internal-api/src/main/java/datadog/trace/util/ChunkedBuffer.java @@ -36,9 +36,9 @@ * so the buffer holds exactly as much as a single array of the same capacity would. The last chunk * is only as long as the remaining capacity, so a small buffer is a single small chunk. * - *

Small payloads stay small. Chunks can start smaller than {@link #CHUNK_SIZE} and double - * up to it (see {@link #ChunkedBuffer(int, int)}), so a buffer whose payloads are usually a few KB - * holds a few KB rather than a whole chunk. + *

Small payloads stay small. Chunks start at {@link #DEFAULT_FIRST_CHUNK_SIZE} and double + * up to {@link #CHUNK_SIZE}, so a buffer whose payloads are usually a few KB holds a few KB rather + * than a whole chunk. A large payload reaches full-size chunks after about 250 KB. * *

Retention. Chunks are allocated on first use. {@link #reset()} keeps the chunks that * held completed data plus half of any surplus kept from the previous reset, so after a burst the @@ -58,6 +58,9 @@ public class ChunkedBuffer { */ public static final int CHUNK_SIZE = 256 * 1024 - 32; + /** The size of the first chunk unless another is given; later chunks double. */ + public static final int DEFAULT_FIRST_CHUNK_SIZE = 4 * 1024; + /** Where writes go once the buffer has failed; reused from the start whenever it fills. */ private static final int SCRATCH_SIZE = 64; @@ -96,7 +99,7 @@ public class ChunkedBuffer { * @param maxBytes the most data the buffer may hold */ public ChunkedBuffer(int maxBytes) { - this(maxBytes, CHUNK_SIZE); + this(maxBytes, DEFAULT_FIRST_CHUNK_SIZE); } /** diff --git a/internal-api/src/test/java/datadog/trace/util/ChunkedBufferTest.java b/internal-api/src/test/java/datadog/trace/util/ChunkedBufferTest.java index dd4cf59b013..c718e6f454a 100644 --- a/internal-api/src/test/java/datadog/trace/util/ChunkedBufferTest.java +++ b/internal-api/src/test/java/datadog/trace/util/ChunkedBufferTest.java @@ -26,14 +26,14 @@ void chunkObjectFitsIn256KiBForEveryArrayHeaderLayout() { @Test void capacityIsExact() { - assertEquals(5 << 20, new ChunkedBuffer(5 << 20).capacity()); - assertEquals(1024, new ChunkedBuffer(1024).capacity()); - assertEquals(0, new ChunkedBuffer(0).capacity()); + assertEquals(5 << 20, new ChunkedBuffer(5 << 20, CHUNK_SIZE).capacity()); + assertEquals(1024, new ChunkedBuffer(1024, CHUNK_SIZE).capacity()); + assertEquals(0, new ChunkedBuffer(0, CHUNK_SIZE).capacity()); } @Test void lastChunkIsShortenedToFitCapacity() { - ChunkedBuffer buffer = new ChunkedBuffer(CHUNK_SIZE + 10); + ChunkedBuffer buffer = new ChunkedBuffer(CHUNK_SIZE + 10, CHUNK_SIZE); buffer.put(new byte[CHUNK_SIZE + 2]); buffer.putLong(5L); assertTrue(buffer.mark()); @@ -46,7 +46,7 @@ void lastChunkIsShortenedToFitCapacity() { @Test void unusedChunkTailsDoNotReduceCapacity() { - ChunkedBuffer buffer = new ChunkedBuffer(TWO_CHUNKS); + ChunkedBuffer buffer = new ChunkedBuffer(TWO_CHUNKS, CHUNK_SIZE); buffer.put(new byte[CHUNK_SIZE - 3]); // leaves 3 unused bytes at the end of the first chunk buffer.putLong(1L); @@ -61,7 +61,7 @@ void unusedChunkTailsDoNotReduceCapacity() { @Test void keptShortChunkIsReplacedWhenAPayloadNeedsMoreOfIt() { - ChunkedBuffer buffer = new ChunkedBuffer(CHUNK_SIZE + 10); + ChunkedBuffer buffer = new ChunkedBuffer(CHUNK_SIZE + 10, CHUNK_SIZE); // no unused tail, so the second chunk only needs to be 10 bytes long buffer.put(new byte[CHUNK_SIZE + 10]); assertTrue(buffer.mark()); @@ -78,7 +78,7 @@ void keptShortChunkIsReplacedWhenAPayloadNeedsMoreOfIt() { @Test void smallBufferIsOneSmallChunk() { - ChunkedBuffer buffer = new ChunkedBuffer(16); + ChunkedBuffer buffer = new ChunkedBuffer(16, CHUNK_SIZE); buffer.putLong(1L); buffer.putLong(2L); assertTrue(buffer.mark()); @@ -91,7 +91,7 @@ void smallBufferIsOneSmallChunk() { @Test void zeroCapacityRejectsEveryMessage() { - ChunkedBuffer buffer = new ChunkedBuffer(0); + ChunkedBuffer buffer = new ChunkedBuffer(0, CHUNK_SIZE); buffer.putInt(1); buffer.put(new byte[10]); assertFalse(buffer.mark()); @@ -101,7 +101,7 @@ void zeroCapacityRejectsEveryMessage() { @Test void allocatesNothingBeforeFirstWrite() { - ChunkedBuffer buffer = new ChunkedBuffer(TWO_CHUNKS); + ChunkedBuffer buffer = new ChunkedBuffer(TWO_CHUNKS, CHUNK_SIZE); assertEquals(0, buffer.allocatedChunks()); buffer.mark(); buffer.reset(); @@ -113,7 +113,7 @@ void allocatesNothingBeforeFirstWrite() { @Test void typedPutsAreBigEndianAcrossChunkBoundaries() { // a mix of widths, repeated enough to cross several chunk boundaries at varying offsets - ChunkedBuffer buffer = new ChunkedBuffer(THREE_CHUNKS); + ChunkedBuffer buffer = new ChunkedBuffer(THREE_CHUNKS, CHUNK_SIZE); ByteBuffer expected = ByteBuffer.allocate(THREE_CHUNKS); int i = 0; while (expected.position() < TWO_CHUNKS + CHUNK_SIZE / 2) { @@ -144,7 +144,7 @@ void typedPutsAreBigEndianAcrossChunkBoundaries() { @Test void primitiveThatDoesNotFitStartsTheNextChunk() { - ChunkedBuffer buffer = new ChunkedBuffer(TWO_CHUNKS); + ChunkedBuffer buffer = new ChunkedBuffer(TWO_CHUNKS, CHUNK_SIZE); buffer.put(new byte[CHUNK_SIZE - 3]); buffer.putLong(0x0102030405060708L); buffer.mark(); @@ -159,7 +159,7 @@ void primitiveThatDoesNotFitStartsTheNextChunk() { @Test void byteArraysSplitAcrossChunks() { - ChunkedBuffer buffer = new ChunkedBuffer(THREE_CHUNKS); + ChunkedBuffer buffer = new ChunkedBuffer(THREE_CHUNKS, CHUNK_SIZE); byte[] data = sequence(CHUNK_SIZE + CHUNK_SIZE / 2); buffer.put((byte) -1); buffer.put(data, 0, data.length); @@ -177,7 +177,7 @@ void byteArraysSplitAcrossChunks() { @Test void byteBufferSplitsAcrossChunks() { - ChunkedBuffer buffer = new ChunkedBuffer(THREE_CHUNKS); + ChunkedBuffer buffer = new ChunkedBuffer(THREE_CHUNKS, CHUNK_SIZE); byte[] data = sequence(CHUNK_SIZE + 100); ByteBuffer src = ByteBuffer.wrap(data); buffer.put(src); @@ -189,7 +189,7 @@ void byteBufferSplitsAcrossChunks() { @Test void emptyMarksAreNotCounted() { - ChunkedBuffer buffer = new ChunkedBuffer(TWO_CHUNKS); + ChunkedBuffer buffer = new ChunkedBuffer(TWO_CHUNKS, CHUNK_SIZE); buffer.mark(); buffer.mark(); assertEquals(0, buffer.messageCount()); @@ -202,7 +202,7 @@ void emptyMarksAreNotCounted() { @Test void rewindDiscardsUnmarkedBytesAcrossChunks() { - ChunkedBuffer buffer = new ChunkedBuffer(THREE_CHUNKS); + ChunkedBuffer buffer = new ChunkedBuffer(THREE_CHUNKS, CHUNK_SIZE); buffer.putInt(42); buffer.mark(); // spill the unfinished message into a second chunk @@ -222,7 +222,7 @@ void rewindDiscardsUnmarkedBytesAcrossChunks() { @Test void rewindBeforeAnyMarkEmptiesTheBuffer() { - ChunkedBuffer buffer = new ChunkedBuffer(TWO_CHUNKS); + ChunkedBuffer buffer = new ChunkedBuffer(TWO_CHUNKS, CHUNK_SIZE); buffer.put(new byte[CHUNK_SIZE + 10]); buffer.rewind(); assertFalse(buffer.isDirty()); @@ -235,7 +235,7 @@ void rewindBeforeAnyMarkEmptiesTheBuffer() { @Test void overflowFailsWithoutThrowingAndMarkRewinds() { - ChunkedBuffer buffer = new ChunkedBuffer(TWO_CHUNKS); + ChunkedBuffer buffer = new ChunkedBuffer(TWO_CHUNKS, CHUNK_SIZE); byte[] first = sequence(CHUNK_SIZE + 10); buffer.put(first); assertTrue(buffer.mark()); @@ -267,7 +267,7 @@ void overflowFailsWithoutThrowingAndMarkRewinds() { @Test void messageLargerThanCapacityFailsFromEmpty() { - ChunkedBuffer buffer = new ChunkedBuffer(CHUNK_SIZE); + ChunkedBuffer buffer = new ChunkedBuffer(CHUNK_SIZE, CHUNK_SIZE); buffer.put(new byte[CHUNK_SIZE + 1]); assertFalse(buffer.mark()); assertFalse(buffer.isDirty()); @@ -276,7 +276,7 @@ void messageLargerThanCapacityFailsFromEmpty() { @Test void writesFillCapacityExactly() { - ChunkedBuffer buffer = new ChunkedBuffer(TWO_CHUNKS); + ChunkedBuffer buffer = new ChunkedBuffer(TWO_CHUNKS, CHUNK_SIZE); buffer.put(new byte[TWO_CHUNKS - 8]); buffer.putLong(1L); assertTrue(buffer.mark()); @@ -289,7 +289,7 @@ void writesFillCapacityExactly() { @Test void rewindAlsoClearsFailure() { - ChunkedBuffer buffer = new ChunkedBuffer(CHUNK_SIZE); + ChunkedBuffer buffer = new ChunkedBuffer(CHUNK_SIZE, CHUNK_SIZE); buffer.put(new byte[CHUNK_SIZE + 1]); buffer.rewind(); buffer.putInt(5); @@ -299,7 +299,7 @@ void rewindAlsoClearsFailure() { @Test void resetKeepsTheChunksTheCompletedMessagesUsed() { - ChunkedBuffer buffer = new ChunkedBuffer(5 * CHUNK_SIZE); + ChunkedBuffer buffer = new ChunkedBuffer(5 * CHUNK_SIZE, CHUNK_SIZE); buffer.put(new byte[2 * CHUNK_SIZE + 1]); buffer.mark(); assertEquals(3, buffer.allocatedChunks()); @@ -312,7 +312,7 @@ void resetKeepsTheChunksTheCompletedMessagesUsed() { @Test void resetReleasesHalfTheSurplusEachTime() { - ChunkedBuffer buffer = new ChunkedBuffer(20 * CHUNK_SIZE); + ChunkedBuffer buffer = new ChunkedBuffer(20 * CHUNK_SIZE, CHUNK_SIZE); buffer.put(new byte[18 * CHUNK_SIZE + 1]); buffer.mark(); buffer.reset(); @@ -330,7 +330,7 @@ void resetReleasesHalfTheSurplusEachTime() { @Test void growthIsNotHalved() { - ChunkedBuffer buffer = new ChunkedBuffer(5 * CHUNK_SIZE); + ChunkedBuffer buffer = new ChunkedBuffer(5 * CHUNK_SIZE, CHUNK_SIZE); buffer.putInt(1); buffer.mark(); buffer.reset(); @@ -344,7 +344,7 @@ void growthIsNotHalved() { @Test void resetIgnoresUnfinishedAndFailedMessages() { - ChunkedBuffer buffer = new ChunkedBuffer(3 * CHUNK_SIZE); + ChunkedBuffer buffer = new ChunkedBuffer(3 * CHUNK_SIZE, CHUNK_SIZE); buffer.putInt(1); buffer.mark(); // an oversized message touches every chunk before failing @@ -355,7 +355,7 @@ void resetIgnoresUnfinishedAndFailedMessages() { @Test void resetReusesRetainedChunks() { - ChunkedBuffer buffer = new ChunkedBuffer(TWO_CHUNKS); + ChunkedBuffer buffer = new ChunkedBuffer(TWO_CHUNKS, CHUNK_SIZE); buffer.put(sequence(CHUNK_SIZE + 10)); buffer.mark(); ByteBuffer before = buffer.completedChunks().get(1); @@ -370,7 +370,7 @@ void resetReusesRetainedChunks() { @Test void completedChunksAreIndependentViews() { - ChunkedBuffer buffer = new ChunkedBuffer(TWO_CHUNKS); + ChunkedBuffer buffer = new ChunkedBuffer(TWO_CHUNKS, CHUNK_SIZE); buffer.putLong(3L); buffer.mark(); @@ -404,6 +404,15 @@ protected void onCapacityExceeded() { assertEquals(8, buffer.completedBytes()); } + @Test + void defaultFirstChunkIsSmall() { + ChunkedBuffer buffer = new ChunkedBuffer(5 << 20); + buffer.putInt(1); + assertTrue(buffer.mark()); + assertEquals( + ChunkedBuffer.DEFAULT_FIRST_CHUNK_SIZE, buffer.completedChunks().get(0).array().length); + } + @Test void chunksStartSmallAndDoubleUpToChunkSize() { ChunkedBuffer buffer = new ChunkedBuffer(4 * CHUNK_SIZE, 4096);