KAFKA-19535: add integration tests for DescribeProducersOptions#brokerId

README.md

@@ -52,13 +52,15 @@ Follow instructions in https://kafka.apache.org/quickstart
 
 ### Running a particular unit/integration test ###
     ./gradlew clients:test --tests RequestResponseTest
+    ./gradlew streams:integration-tests:test --tests RestoreIntegrationTest
 
 ### Repeatedly running a particular unit/integration test with specific times by setting N ###
     N=500; I=0; while [ $I -lt $N ] && ./gradlew clients:test --tests RequestResponseTest --rerun --fail-fast; do (( I=$I+1 )); echo "Completed run: $I"; sleep 1; done
 
 ### Running a particular test method within a unit/integration test ###
     ./gradlew core:test --tests kafka.api.ProducerFailureHandlingTest.testCannotSendToInternalTopic
     ./gradlew clients:test --tests org.apache.kafka.clients.MetadataTest.testTimeToNextUpdate
+    ./gradlew streams:integration-tests:test --tests org.apache.kafka.streams.integration.RestoreIntegrationTest.shouldRestoreNullRecord
 
 ### Running a particular unit/integration test with log4j output ###
 By default, there will be only small number of logs output while testing. You can adjust it by changing the `log4j2.yaml` file in the module's `src/test/resources` directory.

clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/admin/DescribeProducersWithBrokerIdTest.java

@@ -0,0 +1,161 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.admin;
+
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.NotLeaderOrFollowerException;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.apache.kafka.common.test.ClusterInstance;
+import org.apache.kafka.common.test.api.ClusterConfigProperty;
+import org.apache.kafka.common.test.api.ClusterTest;
+import org.apache.kafka.common.test.api.ClusterTestDefaults;
+import org.apache.kafka.test.TestUtils;
+
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG;
+import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG;
+import static org.apache.kafka.server.config.ServerLogConfigs.AUTO_CREATE_TOPICS_ENABLE_CONFIG;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+
+@ClusterTestDefaults(
+    brokers = 4,
+    serverProperties = {
+        @ClusterConfigProperty(key = AUTO_CREATE_TOPICS_ENABLE_CONFIG, value = "false"),
+        @ClusterConfigProperty(key = OFFSETS_TOPIC_PARTITIONS_CONFIG, value = "1"),
+        @ClusterConfigProperty(key = GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, value = "0")
+    }
+)
+class DescribeProducersWithBrokerIdTest {
+    private static final String TOPIC_NAME = "test-topic";
+    private static final int NUM_PARTITIONS = 3;
+    private static final short REPLICATION_FACTOR = 3;
+
+    private final ClusterInstance clusterInstance;
+    public DescribeProducersWithBrokerIdTest(ClusterInstance clusterInstance) {
+        this.clusterInstance = clusterInstance;
+    }
+
+    private static void sendTestRecords(Producer<Object, Object> producer) {
+        for (int partition = 0; partition < NUM_PARTITIONS; partition++) {
+            producer.send(new ProducerRecord<>(TOPIC_NAME, partition, "key-" + partition, "value-" + partition));
+        }
+        producer.flush();
+    }
+
+    @ClusterTest
+    void testDescribeProducersDefaultRoutesToLeader() throws Exception {
+        clusterInstance.createTopic(TOPIC_NAME, NUM_PARTITIONS, REPLICATION_FACTOR);
+
+        try (var producer = clusterInstance.producer(Map.of(
+                ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class,
+                ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class));
+             var admin = clusterInstance.admin()) {
+
+            sendTestRecords(producer);
+
+            var topicPartition = new TopicPartition(TOPIC_NAME, 0);
+            var leaderBrokerId = clusterInstance.getLeaderBrokerId(topicPartition);
+
+            var stateWithExplicitLeader = admin.describeProducers(List.of(topicPartition), 
+                    new DescribeProducersOptions().brokerId(leaderBrokerId))
+                    .partitionResult(topicPartition).get();
+
+            var stateWithDefaultRouting = admin.describeProducers(List.of(topicPartition))
+                    .partitionResult(topicPartition).get();
+
+            assertNotNull(stateWithDefaultRouting);
+            assertFalse(stateWithDefaultRouting.activeProducers().isEmpty());
+            assertEquals(stateWithExplicitLeader.activeProducers(), stateWithDefaultRouting.activeProducers());
+        }
+    }
+
+    @ClusterTest
+    void testDescribeProducersFromFollower() throws Exception {
+        clusterInstance.createTopic(TOPIC_NAME, NUM_PARTITIONS, REPLICATION_FACTOR);
+
+        try (var producer = clusterInstance.producer(Map.of(
+                ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class,
+                ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class));
+             var admin = clusterInstance.admin()) {
+
+            sendTestRecords(producer);
+
+            var topicPartition = new TopicPartition(TOPIC_NAME, 0);
+            var topicDescription = admin.describeTopics(List.of(TOPIC_NAME)).allTopicNames().get().get(TOPIC_NAME);
+            var replicaBrokerIds = topicDescription.partitions().get(0).replicas().stream()
+                    .map(Node::id)
+                    .toList();
+
+            var leaderBrokerId = clusterInstance.getLeaderBrokerId(topicPartition);
+            var followerBrokerId = replicaBrokerIds.stream()
+                    .filter(id -> !id.equals(leaderBrokerId))
+                    .findFirst()
+                    .orElseThrow(() -> new IllegalStateException("No follower found"));
+
+            var followerState = admin.describeProducers(List.of(topicPartition), 
+                    new DescribeProducersOptions().brokerId(followerBrokerId))
+                    .partitionResult(topicPartition).get();
+            var leaderState = admin.describeProducers(List.of(topicPartition))
+                    .partitionResult(topicPartition).get();
+
+            assertNotNull(followerState);
+            assertFalse(followerState.activeProducers().isEmpty());
+
+            var followerProducerId = followerState.activeProducers().iterator().next().producerId();
+            var leaderProducerId = leaderState.activeProducers().iterator().next().producerId();
+            assertEquals(leaderProducerId, followerProducerId);
+        }
+    }
+
+    @ClusterTest
+    void testDescribeProducersWithInvalidBrokerId() throws Exception {
+        clusterInstance.createTopic(TOPIC_NAME, NUM_PARTITIONS, REPLICATION_FACTOR);
+
+        try (var producer = clusterInstance.producer(Map.of(
+                ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class,
+                ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class));
+             var admin = clusterInstance.admin()) {
+
+            sendTestRecords(producer);
+
+            var topicPartition = new TopicPartition(TOPIC_NAME, 0);
+            var topicDescription = admin.describeTopics(List.of(TOPIC_NAME)).allTopicNames().get().get(TOPIC_NAME);
+            var replicaBrokerIds = topicDescription.partitions().get(0).replicas().stream()
+                    .map(Node::id)
+                    .toList();
+
+            var nonReplicaBrokerId = clusterInstance.brokerIds().stream()
+                    .filter(id -> !replicaBrokerIds.contains(id))
+                    .findFirst()
+                    .orElseThrow(() -> new IllegalStateException("No non-replica broker found"));
+
+            TestUtils.assertFutureThrows(NotLeaderOrFollowerException.class, 
+                    admin.describeProducers(List.of(topicPartition), 
+                            new DescribeProducersOptions().brokerId(nonReplicaBrokerId))
+                            .partitionResult(topicPartition)); 
+        }
+    }
+}

clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/PlaintextConsumerSubscriptionTest.java

@@ -576,7 +576,7 @@ public void testClassicConsumerSubscribeInvalidTopicCanUnsubscribe() throws Inte
     }
 
     @ClusterTest
-    public void testAsyncConsumerClassicConsumerSubscribeInvalidTopicCanUnsubscribe() throws InterruptedException {
+    public void testAsyncConsumerSubscribeInvalidTopicCanUnsubscribe() throws InterruptedException {
         testSubscribeInvalidTopicCanUnsubscribe(GroupProtocol.CONSUMER);
     }
 

clients/src/main/java/org/apache/kafka/clients/NetworkClient.java

@@ -1051,9 +1051,9 @@ private void handleApiVersionsResponse(List<ClientResponse> responses,
             apiVersionsResponse.data().finalizedFeaturesEpoch());
         apiVersions.update(node, nodeVersionInfo);
         this.connectionStates.ready(node);
-        log.debug("Node {} has finalized features epoch: {}, finalized features: {}, supported features: {}, ZK migration ready: {}, API versions: {}.",
+        log.debug("Node {} has finalized features epoch: {}, finalized features: {}, supported features: {}, API versions: {}.",
                 node, apiVersionsResponse.data().finalizedFeaturesEpoch(), apiVersionsResponse.data().finalizedFeatures(),
-                apiVersionsResponse.data().supportedFeatures(), apiVersionsResponse.data().zkMigrationReady(), nodeVersionInfo);
+                apiVersionsResponse.data().supportedFeatures(), nodeVersionInfo);
     }
 
     /**

clients/src/main/java/org/apache/kafka/clients/admin/AlterConfigOp.java

@@ -26,6 +26,20 @@
 
 /**
  * A class representing an alter configuration entry containing name, value and operation type.
+ * <p>
+ * <b>Note for Broker Logger Configuration:</b><br>
+ * When altering broker logger levels (using {@link org.apache.kafka.common.config.ConfigResource.Type#BROKER_LOGGER}),
+ * it is strongly recommended to use log level constants from {@link org.apache.kafka.common.config.LogLevelConfig} instead of string literals.
+ * This ensures compatibility with Kafka's log level validation and avoids potential configuration errors.
+ * <p>
+ * Example:
+ * <pre>
+ * Recommended approach:
+ * new AlterConfigOp(new ConfigEntry(loggerName, LogLevelConfig.DEBUG_LOG_LEVEL), OpType.SET)
+ *
+ * Avoid this:
+ * new AlterConfigOp(new ConfigEntry(loggerName, "DEBUG"), OpType.SET)
+ * </pre>
  */
 public class AlterConfigOp {
 

clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java

@@ -2334,7 +2334,7 @@ private Map<String, KafkaFuture<TopicDescription>> handleDescribeTopicsByNamesWi
         }
 
         // First, we need to retrieve the node info.
-        DescribeClusterResult clusterResult = describeCluster();
+        DescribeClusterResult clusterResult = describeCluster(new DescribeClusterOptions().timeoutMs(options.timeoutMs()));
         clusterResult.nodes().whenComplete(
             (nodes, exception) -> {
                 if (exception != null) {
@@ -5154,6 +5154,8 @@ private static long getOffsetFromSpec(OffsetSpec offsetSpec) {
             return ListOffsetsRequest.EARLIEST_LOCAL_TIMESTAMP;
         } else if (offsetSpec instanceof OffsetSpec.LatestTieredSpec) {
             return ListOffsetsRequest.LATEST_TIERED_TIMESTAMP;
+        } else if (offsetSpec instanceof OffsetSpec.EarliestPendingUploadSpec) {
+            return ListOffsetsRequest.EARLIEST_PENDING_UPLOAD_TIMESTAMP;
         }
         return ListOffsetsRequest.LATEST_TIMESTAMP;
     }

clients/src/main/java/org/apache/kafka/clients/admin/OffsetSpec.java

@@ -28,6 +28,7 @@ public static class LatestSpec extends OffsetSpec { }
     public static class MaxTimestampSpec extends OffsetSpec { }
     public static class EarliestLocalSpec extends OffsetSpec { }
     public static class LatestTieredSpec extends OffsetSpec { }
+    public static class EarliestPendingUploadSpec extends OffsetSpec { }
     public static class TimestampSpec extends OffsetSpec {
         private final long timestamp;
 
@@ -91,4 +92,13 @@ public static OffsetSpec earliestLocal() {
     public static OffsetSpec latestTiered() {
         return new LatestTieredSpec();
     }
+
+    /**
+     * Used to retrieve the earliest offset of records that are pending upload to remote storage.
+     * <br/>
+     * Note: When tiered storage is not enabled, we will return unknown offset.
+     */
+    public static OffsetSpec earliestPendingUpload() {
+        return new EarliestPendingUploadSpec();
+    }
 }

clients/src/main/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandler.java

@@ -103,12 +103,17 @@ ListOffsetsRequest.Builder buildBatchedRequest(int brokerId, Set<TopicPartition>
             .stream()
             .anyMatch(key -> offsetTimestampsByPartition.get(key) == ListOffsetsRequest.LATEST_TIERED_TIMESTAMP);
 
+        boolean requireEarliestPendingUploadTimestamp = keys
+            .stream()
+            .anyMatch(key -> offsetTimestampsByPartition.get(key) == ListOffsetsRequest.EARLIEST_PENDING_UPLOAD_TIMESTAMP);
+
         int timeoutMs = options.timeoutMs() != null ? options.timeoutMs() : defaultApiTimeoutMs;
         return ListOffsetsRequest.Builder.forConsumer(true,
                         options.isolationLevel(),
                         supportsMaxTimestamp,
                         requireEarliestLocalTimestamp,
-                        requireTieredStorageTimestamp)
+                        requireTieredStorageTimestamp,
+                        requireEarliestPendingUploadTimestamp)
                 .setTargetTimes(new ArrayList<>(topicsByName.values()))
                 .setTimeoutMs(timeoutMs);
     }

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

@@ -47,6 +47,8 @@ public class ListOffsetsRequest extends AbstractRequest {
 
     public static final long LATEST_TIERED_TIMESTAMP = -5L;
 
+    public static final long EARLIEST_PENDING_UPLOAD_TIMESTAMP = -6L;
+
     public static final int CONSUMER_REPLICA_ID = -1;
     public static final int DEBUGGING_REPLICA_ID = -2;
 
@@ -58,16 +60,19 @@ public static class Builder extends AbstractRequest.Builder<ListOffsetsRequest>
 
         public static Builder forConsumer(boolean requireTimestamp,
                                           IsolationLevel isolationLevel) {
-            return forConsumer(requireTimestamp, isolationLevel, false, false, false);
+            return forConsumer(requireTimestamp, isolationLevel, false, false, false, false);
         }
 
         public static Builder forConsumer(boolean requireTimestamp,
                                           IsolationLevel isolationLevel,
                                           boolean requireMaxTimestamp,
                                           boolean requireEarliestLocalTimestamp,
-                                          boolean requireTieredStorageTimestamp) {
+                                          boolean requireTieredStorageTimestamp,
+                                          boolean requireEarliestPendingUploadTimestamp) {
             short minVersion = ApiKeys.LIST_OFFSETS.oldestVersion();
-            if (requireTieredStorageTimestamp)
+            if (requireEarliestPendingUploadTimestamp)
+                minVersion = 11;
+            else if (requireTieredStorageTimestamp)
                 minVersion = 9;
             else if (requireEarliestLocalTimestamp)
                 minVersion = 8;

clients/src/main/resources/common/message/ListOffsetsRequest.json

@@ -40,7 +40,9 @@
   // Version 9 enables listing offsets by last tiered offset (KIP-1005).
   //
   // Version 10 enables async remote list offsets support (KIP-1075)
-  "validVersions": "1-10",
+  //
+  // Version 11 enables listing offsets by earliest pending upload offset (KIP-1023)
+  "validVersions": "1-11",
   "flexibleVersions": "6+",
   "latestVersionUnstable": false,
   "fields": [

clients/src/main/resources/common/message/ListOffsetsResponse.json

@@ -40,7 +40,9 @@
   // Version 9 enables listing offsets by last tiered offset (KIP-1005).
   //
   // Version 10 enables async remote list offsets support (KIP-1075)
-  "validVersions": "1-10",
+  //
+  // Version 11 enables listing offsets by earliest pending upload offset (KIP-1023)
+  "validVersions": "1-11",
   "flexibleVersions": "6+",
   "fields": [
     { "name": "ThrottleTimeMs", "type": "int32", "versions": "2+", "ignorable": true,