Merge branch 'apache:trunk' into storage

Author: jiafu1115

build.gradle

@@ -2467,6 +2467,7 @@ project(':storage') {
     testImplementation project(':core')
     testImplementation project(':core').sourceSets.test.output
     testImplementation testFixtures(project(':storage:storage-api'))
+    testImplementation project(':metadata')
     testImplementation project(':test-common:test-common-internal-api')
     testImplementation project(':test-common:test-common-runtime')
     testImplementation project(':test-common:test-common-util')
@@ -2625,6 +2626,12 @@ project(':tools') {
 
   configurations {
     releaseOnly
+    // ApacheDS 2.0.0-M24 pulls in the stale bcprov-jdk15on:1.56, which ships the same
+    // RosstandartObjectIdentifiers class as the modern bcprov-jdk18on we already depend on
+    // but is missing fields referenced by bcpkix-jdk18on:1.84. When IntelliJ's test runner
+    // orders the old JAR first, BC provider registration fails with NoSuchFieldError. Drop it.
+    testCompileClasspath.exclude group: 'org.bouncycastle', module: 'bcprov-jdk15on'
+    testRuntimeClasspath.exclude group: 'org.bouncycastle', module: 'bcprov-jdk15on'
   }
 
   dependencies {

checkstyle/import-control-storage.xml

@@ -118,6 +118,7 @@
         <allow pkg="org.apache.kafka.tiered.storage" />
 
         <allow pkg="kafka.api" />
+        <allow pkg="kafka.integration" />
         <allow pkg="kafka.log" />
         <allow pkg="kafka.server" />
         <allow pkg="kafka.utils" />

clients/src/main/java/org/apache/kafka/common/requests/PushTelemetryRequest.java

@@ -85,10 +85,10 @@ public String metricsContentType() {
         return OTLP_CONTENT_TYPE;
     }
 
-    public ByteBuffer metricsData() {
+    public ByteBuffer metricsData(int maxDecompressedBytes) {
         CompressionType cType = CompressionType.forId(this.data.compressionType());
         return (cType == CompressionType.NONE) ?
-            this.data.metrics() : ClientTelemetryUtils.decompress(this.data.metrics(), cType);
+            this.data.metrics() : ClientTelemetryUtils.decompress(this.data.metrics(), cType, maxDecompressedBytes);
     }
 
     public static PushTelemetryRequest parse(Readable readable, short version) {

clients/src/main/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryUtils.java

@@ -19,6 +19,7 @@
 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.internal.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:
      *
@@ -212,13 +215,18 @@ public static ByteBuffer compress(MetricsData metrics, CompressionType compressi
         }
     }
 
-    public static ByteBuffer decompress(ByteBuffer metrics, CompressionType compressionType) {
+    public static ByteBuffer decompress(ByteBuffer metrics, CompressionType compressionType, int maxDecompressedBytes) {
         Compression compression = Compression.of(compressionType).build();
         try (InputStream in = compression.wrapForInput(metrics, RecordBatch.CURRENT_MAGIC_VALUE, BufferSupplier.create());
             ByteBufferOutputStream out = new ByteBufferOutputStream(512)) {
-            byte[] bytes = new byte[metrics.limit() * 2];
+            byte[] bytes = new byte[Math.min(metrics.limit() * 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();

clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java

@@ -1407,8 +1407,9 @@ public void testNoCommittedOffsets(GroupProtocol groupProtocol) {
         assertNull(committed.get(tp1));
     }
 
+    // NOTE: the rebalance flow in prepareRebalance is specific to the CLASSIC consumer.
     @ParameterizedTest
-    @EnumSource(GroupProtocol.class)
+    @EnumSource(value = GroupProtocol.class, names = "CLASSIC")
     public void testAutoCommitSentBeforePositionUpdate(GroupProtocol groupProtocol) {
         ConsumerMetadata metadata = createMetadata(subscription);
         MockClient client = new MockClient(time, metadata);

clients/src/test/java/org/apache/kafka/common/requests/PushTelemetryRequestTest.java

@@ -62,7 +62,7 @@ public void testMetricsDataCompression(CompressionType compressionType) throws I
         MetricsData metricsData = getMetricsData();
         PushTelemetryRequest req = getPushTelemetryRequest(metricsData, compressionType);
 
-        ByteBuffer receivedMetricsBuffer = req.metricsData();
+        ByteBuffer receivedMetricsBuffer = req.metricsData(1024 * 1024);
         assertNotNull(receivedMetricsBuffer);
         assertTrue(receivedMetricsBuffer.capacity() > 0);
 

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.internal.CompressionType;
 import org.apache.kafka.common.utils.Utils;
@@ -154,7 +155,7 @@ public void testCompressDecompress(CompressionType compressionType) throws IOExc
         } else {
             assertArrayEquals(raw, Utils.toArray(compressed));
         }
-        ByteBuffer decompressed = ClientTelemetryUtils.decompress(compressed, compressionType);
+        ByteBuffer decompressed = ClientTelemetryUtils.decompress(compressed, compressionType, 1024 * 1024);
         assertNotNull(decompressed);
         byte[] actualResult = Utils.toArray(decompressed);
         assertArrayEquals(raw, actualResult);
@@ -191,4 +192,31 @@ private MetricsData getMetricsData() {
 
         return builder.build();
     }
+
+    @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
+        MetricsData metricsData = getMetricsData();
+        ByteBuffer compressed = ClientTelemetryUtils.compress(metricsData, CompressionType.GZIP);
+        byte[] raw = metricsData.toByteArray();
+
+        // Set limit smaller than the actual decompressed size
+        int smallLimit = raw.length - 1;
+        TelemetryTooLargeException ex = assertThrows(TelemetryTooLargeException.class,
+            () -> ClientTelemetryUtils.decompress(compressed.duplicate(), CompressionType.GZIP, smallLimit));
+        assertTrue(ex.getMessage().contains("Decompressed telemetry metrics exceed maximum allowed size: " + smallLimit));
+    }
+
+    @Test
+    public void testDecompressWithPayloadSizeSucceeds() throws IOException {
+        MetricsData metricsData = getMetricsData();
+        byte[] raw = metricsData.toByteArray();
+        ByteBuffer compressed = ClientTelemetryUtils.compress(metricsData, CompressionType.GZIP);
+
+        // Set limit to exact limit prior compression.
+        ByteBuffer result = ClientTelemetryUtils.decompress(compressed, CompressionType.GZIP, raw.length);
+        assertNotNull(result);
+        assertArrayEquals(raw, Utils.toArray(result));
+    }
 }

connect/api/src/main/java/org/apache/kafka/connect/source/TransactionContext.java

@@ -43,7 +43,7 @@ public interface TransactionContext {
     /**
      * Requests a transaction abort after the next batch of records from {@link SourceTask#poll()}. All of
      * the records in that transaction will be discarded and will not appear in a committed transaction.
-     * However, offsets for that transaction will still be committed so than the records in that transaction
+     * However, offsets for that transaction will still be committed so that the records in that transaction
      * are not reprocessed. If the data should instead be reprocessed, the task should not invoke this method
      * and should instead throw an exception.
      */

connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationBaseTest.java

@@ -287,6 +287,8 @@ public void shutdownClusters() throws Exception {
             } finally {
                 Exit.resetExitProcedure();
                 Exit.resetHaltProcedure();
+                // Shared by all CollectAllMetricsReporter instances; avoid cross-test accumulation.
+                CollectAllMetricsReporter.METRICS.clear();
             }
         }
     }
@@ -1628,5 +1630,10 @@ public void init(List<KafkaMetric> metrics) {
         public void metricChange(KafkaMetric metric) {
             METRICS.put(metric.metricName(), metric);
         }
+
+        @Override
+        public void metricRemoval(KafkaMetric metric) {
+            METRICS.remove(metric.metricName());
+        }
     }
 }

core/src/main/java/kafka/server/builders/KafkaApisBuilder.java

@@ -19,7 +19,6 @@
 
 import kafka.coordinator.transaction.TransactionCoordinator;
 import kafka.network.RequestChannel;
-import kafka.server.AutoTopicCreationManager;
 import kafka.server.ForwardingManager;
 import kafka.server.KafkaApis;
 import kafka.server.KafkaConfig;
@@ -37,6 +36,7 @@
 import org.apache.kafka.metadata.MetadataCache;
 import org.apache.kafka.security.DelegationTokenManager;
 import org.apache.kafka.server.ApiVersionManager;
+import org.apache.kafka.server.AutoTopicCreationManager;
 import org.apache.kafka.server.ClientMetricsManager;
 import org.apache.kafka.server.FetchManager;
 import org.apache.kafka.server.authorizer.Authorizer;