This is an automated email from the ASF dual-hosted git repository.
AndrewJSchofield pushed a commit to branch 3.9
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/3.9 by this push:
new 1c85bc56312 MINOR: Fixed metrics decompression (#22327)
1c85bc56312 is described below
commit 1c85bc563121b0135e948328f54d939eaae3740b
Author: Apoorv Mittal <[email protected]>
AuthorDate: Tue May 19 18:34:06 2026 +0100
MINOR: Fixed metrics decompression (#22327)
Minor fix for fixing decompression.
Reviewers: Andrew Schofield <[email protected]>
---
.../common/requests/PushTelemetryRequest.java | 4 ++--
.../telemetry/internals/ClientTelemetryUtils.java | 12 ++++++++--
.../common/requests/PushTelemetryRequestTest.java | 2 +-
.../internals/ClientTelemetryUtilsTest.java | 28 +++++++++++++++++++++-
.../apache/kafka/server/ClientMetricsManager.java | 2 +-
.../metrics/ClientMetricsReceiverPlugin.java | 8 +++----
.../metrics/DefaultClientTelemetryPayload.java | 4 ++--
.../kafka/server/ClientMetricsManagerTest.java | 2 +-
.../metrics/ClientMetricsReceiverPluginTest.java | 2 +-
9 files changed, 49 insertions(+), 15 deletions(-)
diff --git
a/clients/src/main/java/org/apache/kafka/common/requests/PushTelemetryRequest.java
b/clients/src/main/java/org/apache/kafka/common/requests/PushTelemetryRequest.java
index 94e029d4db1..0e3fa6d1ba3 100644
---
a/clients/src/main/java/org/apache/kafka/common/requests/PushTelemetryRequest.java
+++
b/clients/src/main/java/org/apache/kafka/common/requests/PushTelemetryRequest.java
@@ -85,10 +85,10 @@ public class PushTelemetryRequest extends AbstractRequest {
return OTLP_CONTENT_TYPE;
}
- public ByteBuffer metricsData() {
+ public ByteBuffer metricsData(int maxDecompressedBytes) {
CompressionType cType =
CompressionType.forId(this.data.compressionType());
return (cType == CompressionType.NONE) ?
- ByteBuffer.wrap(this.data.metrics()) :
ClientTelemetryUtils.decompress(this.data.metrics(), cType);
+ ByteBuffer.wrap(this.data.metrics()) :
ClientTelemetryUtils.decompress(this.data.metrics(), cType,
maxDecompressedBytes);
}
public static PushTelemetryRequest parse(ByteBuffer buffer, short version)
{
diff --git
a/clients/src/main/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryUtils.java
b/clients/src/main/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryUtils.java
index c78fc4eeb49..179dd8e92a1 100644
---
a/clients/src/main/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryUtils.java
+++
b/clients/src/main/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryUtils.java
@@ -19,6 +19,7 @@ package org.apache.kafka.common.telemetry.internals;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.compress.Compression;
+import org.apache.kafka.common.errors.TelemetryTooLargeException;
import org.apache.kafka.common.metrics.MetricsContext;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.record.CompressionType;
@@ -52,6 +53,8 @@ public class ClientTelemetryUtils {
public static final Predicate<? super MetricKeyable> SELECTOR_ALL_METRICS
= k -> true;
+ private static final int DECOMPRESS_READ_BUFFER_BYTES = 8 * 1024;
+
/**
* Examine the response data and handle different error code accordingly:
*
@@ -203,14 +206,19 @@ public class ClientTelemetryUtils {
}
}
- public static ByteBuffer decompress(byte[] metrics, CompressionType
compressionType) {
+ public static ByteBuffer decompress(byte[] metrics, CompressionType
compressionType, int maxDecompressedBytes) {
ByteBuffer data = ByteBuffer.wrap(metrics);
Compression compression = Compression.of(compressionType).build();
try (InputStream in = compression.wrapForInput(data,
RecordBatch.CURRENT_MAGIC_VALUE, BufferSupplier.create());
ByteBufferOutputStream out = new ByteBufferOutputStream(512)) {
- byte[] bytes = new byte[data.capacity() * 2];
+ byte[] bytes = new byte[Math.min(data.capacity() * 2,
DECOMPRESS_READ_BUFFER_BYTES)];
int nRead;
+ int totalRead = 0;
while ((nRead = in.read(bytes, 0, bytes.length)) != -1) {
+ totalRead += nRead;
+ if (totalRead > maxDecompressedBytes) {
+ throw new TelemetryTooLargeException("Decompressed
telemetry metrics exceed maximum allowed size: " + maxDecompressedBytes);
+ }
out.write(bytes, 0, nRead);
}
out.buffer().flip();
diff --git
a/clients/src/test/java/org/apache/kafka/common/requests/PushTelemetryRequestTest.java
b/clients/src/test/java/org/apache/kafka/common/requests/PushTelemetryRequestTest.java
index 0727af355e8..6ddc967aa2f 100644
---
a/clients/src/test/java/org/apache/kafka/common/requests/PushTelemetryRequestTest.java
+++
b/clients/src/test/java/org/apache/kafka/common/requests/PushTelemetryRequestTest.java
@@ -61,7 +61,7 @@ public class PushTelemetryRequestTest {
MetricsData metricsData = getMetricsData();
PushTelemetryRequest req = getPushTelemetryRequest(metricsData,
compressionType);
- ByteBuffer receivedMetricsBuffer = req.metricsData();
+ ByteBuffer receivedMetricsBuffer = req.metricsData(1024 * 1024);
assertNotNull(receivedMetricsBuffer);
assertTrue(receivedMetricsBuffer.capacity() > 0);
diff --git
a/clients/src/test/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryUtilsTest.java
b/clients/src/test/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryUtilsTest.java
index d3816bfedaa..c5e70426294 100644
---
a/clients/src/test/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryUtilsTest.java
+++
b/clients/src/test/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryUtilsTest.java
@@ -17,6 +17,7 @@
package org.apache.kafka.common.telemetry.internals;
import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.errors.TelemetryTooLargeException;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.record.CompressionType;
import org.apache.kafka.common.utils.Utils;
@@ -134,9 +135,34 @@ public class ClientTelemetryUtilsTest {
} else {
assertArrayEquals(testString, compressed);
}
- ByteBuffer decompressed = ClientTelemetryUtils.decompress(compressed,
compressionType);
+ ByteBuffer decompressed = ClientTelemetryUtils.decompress(compressed,
compressionType, 1024 * 1024);
assertNotNull(decompressed);
byte[] actualResult = Utils.toArray(decompressed);
assertArrayEquals(testString, actualResult);
}
+
+ @Test
+ public void testDecompressExceedingMaxSizeThrows() throws IOException {
+ // Compress a large payload using the existing compress API (via
MetricsData)
+ // then verify decompression with a small limit throws
+ byte[] testString =
"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa".getBytes(StandardCharsets.UTF_8);
+ byte[] compressed = ClientTelemetryUtils.compress(testString,
CompressionType.GZIP);
+
+ // Set limit smaller than the actual decompressed size
+ int smallLimit = compressed.length - 1;
+ TelemetryTooLargeException ex =
assertThrows(TelemetryTooLargeException.class,
+ () -> ClientTelemetryUtils.decompress(compressed,
CompressionType.GZIP, smallLimit));
+ assertTrue(ex.getMessage().contains("Decompressed telemetry metrics
exceed maximum allowed size: " + smallLimit));
+ }
+
+ @Test
+ public void testDecompressWithPayloadSizeSucceeds() throws IOException {
+ byte[] testString =
"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa".getBytes(StandardCharsets.UTF_8);
+ byte[] compressed = ClientTelemetryUtils.compress(testString,
CompressionType.GZIP);
+
+ // Set limit to exact limit prior compression.
+ ByteBuffer result = ClientTelemetryUtils.decompress(compressed,
CompressionType.GZIP, testString.length);
+ assertNotNull(result);
+ assertArrayEquals(testString, Utils.toArray(result));
+ }
}
\ No newline at end of file
diff --git
a/server/src/main/java/org/apache/kafka/server/ClientMetricsManager.java
b/server/src/main/java/org/apache/kafka/server/ClientMetricsManager.java
index c3f76ae9dee..5ac25e28eb5 100644
--- a/server/src/main/java/org/apache/kafka/server/ClientMetricsManager.java
+++ b/server/src/main/java/org/apache/kafka/server/ClientMetricsManager.java
@@ -209,7 +209,7 @@ public class ClientMetricsManager implements AutoCloseable {
if (metrics != null && metrics.length > 0) {
try {
long exportTimeStartMs = time.hiResClockMs();
- receiverPlugin.exportMetrics(requestContext, request);
+ receiverPlugin.exportMetrics(requestContext, request,
clientTelemetryMaxBytes);
clientMetricsStats.recordPluginExport(clientInstanceId,
time.hiResClockMs() - exportTimeStartMs);
} catch (Throwable exception) {
clientMetricsStats.recordPluginErrorCount(clientInstanceId);
diff --git
a/server/src/main/java/org/apache/kafka/server/metrics/ClientMetricsReceiverPlugin.java
b/server/src/main/java/org/apache/kafka/server/metrics/ClientMetricsReceiverPlugin.java
index c608510350a..137cff0c1b3 100644
---
a/server/src/main/java/org/apache/kafka/server/metrics/ClientMetricsReceiverPlugin.java
+++
b/server/src/main/java/org/apache/kafka/server/metrics/ClientMetricsReceiverPlugin.java
@@ -44,12 +44,12 @@ public class ClientMetricsReceiverPlugin {
receivers.add(receiver);
}
- public DefaultClientTelemetryPayload getPayLoad(PushTelemetryRequest
request) {
- return new DefaultClientTelemetryPayload(request);
+ public DefaultClientTelemetryPayload getPayLoad(PushTelemetryRequest
request, int maxDecompressedBytes) {
+ return new DefaultClientTelemetryPayload(request,
maxDecompressedBytes);
}
- public void exportMetrics(RequestContext context, PushTelemetryRequest
request) {
- DefaultClientTelemetryPayload payload = getPayLoad(request);
+ public void exportMetrics(RequestContext context, PushTelemetryRequest
request, int maxDecompressedBytes) {
+ DefaultClientTelemetryPayload payload = getPayLoad(request,
maxDecompressedBytes);
for (ClientTelemetryReceiver receiver : receivers) {
receiver.exportMetrics(context, payload);
}
diff --git
a/server/src/main/java/org/apache/kafka/server/metrics/DefaultClientTelemetryPayload.java
b/server/src/main/java/org/apache/kafka/server/metrics/DefaultClientTelemetryPayload.java
index f90f473b1ed..0db7fcb6de1 100644
---
a/server/src/main/java/org/apache/kafka/server/metrics/DefaultClientTelemetryPayload.java
+++
b/server/src/main/java/org/apache/kafka/server/metrics/DefaultClientTelemetryPayload.java
@@ -32,11 +32,11 @@ public class DefaultClientTelemetryPayload implements
ClientTelemetryPayload {
private final String metricsContentType;
private final ByteBuffer metricsData;
- DefaultClientTelemetryPayload(PushTelemetryRequest request) {
+ DefaultClientTelemetryPayload(PushTelemetryRequest request, int
maxDecompressedBytes) {
this.clientInstanceId = request.data().clientInstanceId();
this.isClientTerminating = request.data().terminating();
this.metricsContentType = request.metricsContentType();
- this.metricsData = request.metricsData();
+ this.metricsData = request.metricsData(maxDecompressedBytes);
}
@Override
diff --git
a/server/src/test/java/org/apache/kafka/server/ClientMetricsManagerTest.java
b/server/src/test/java/org/apache/kafka/server/ClientMetricsManagerTest.java
index de6aad8f204..ae3fc6eac3e 100644
--- a/server/src/test/java/org/apache/kafka/server/ClientMetricsManagerTest.java
+++ b/server/src/test/java/org/apache/kafka/server/ClientMetricsManagerTest.java
@@ -1076,7 +1076,7 @@ public class ClientMetricsManagerTest {
@Test
public void testPushTelemetryPluginException() throws Exception {
ClientMetricsReceiverPlugin receiverPlugin =
Mockito.mock(ClientMetricsReceiverPlugin.class);
- Mockito.doThrow(new RuntimeException("test
exception")).when(receiverPlugin).exportMetrics(Mockito.any(), Mockito.any());
+ Mockito.doThrow(new RuntimeException("test
exception")).when(receiverPlugin).exportMetrics(Mockito.any(), Mockito.any(),
Mockito.anyInt());
try (
Metrics kafkaMetrics = new Metrics();
diff --git
a/server/src/test/java/org/apache/kafka/server/metrics/ClientMetricsReceiverPluginTest.java
b/server/src/test/java/org/apache/kafka/server/metrics/ClientMetricsReceiverPluginTest.java
index 35f9a41b47d..225811d726d 100644
---
a/server/src/test/java/org/apache/kafka/server/metrics/ClientMetricsReceiverPluginTest.java
+++
b/server/src/test/java/org/apache/kafka/server/metrics/ClientMetricsReceiverPluginTest.java
@@ -53,7 +53,7 @@ public class ClientMetricsReceiverPluginTest {
byte[] metrics = "test-metrics".getBytes(StandardCharsets.UTF_8);
clientMetricsReceiverPlugin.exportMetrics(ClientMetricsTestUtils.requestContext(),
- new PushTelemetryRequest.Builder(new
PushTelemetryRequestData().setMetrics(metrics), true).build());
+ new PushTelemetryRequest.Builder(new
PushTelemetryRequestData().setMetrics(metrics), true).build(),
MetricConfigs.CLIENT_TELEMETRY_MAX_BYTES_DEFAULT);
assertEquals(1, telemetryReceiver.exportMetricsInvokedCount);
assertEquals(1, telemetryReceiver.metricsData.size());
