init commit

Author: TaiJuWu

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

@@ -2993,6 +2993,10 @@ public void testRenewAcknowledgementOnCommitSync() {
             shareConsumer.subscribe(List.of(tp.topic()));
             ConsumerRecords<byte[], byte[]> records = waitedPoll(shareConsumer, 2500L, 10);
             assertEquals(10, records.count());
+            assertEquals(Optional.of(15000), shareConsumer.acquisitionLockTimeoutMs());
+
+            // The updated acquisition lock timeout is only applied when the next poll is called.
+            alterShareRecordLockDurationMs("group1", 25000);
 
             int count = 0;
             Map<TopicIdPartition, Optional<KafkaException>> result;
@@ -3004,13 +3008,15 @@ public void testRenewAcknowledgementOnCommitSync() {
                 }
                 result = shareConsumer.commitSync();
                 assertEquals(1, result.size());
+                assertEquals(Optional.of(15000), shareConsumer.acquisitionLockTimeoutMs());
                 assertEquals(Optional.empty(), result.get(new TopicIdPartition(tpId, tp.partition(), tp.topic())));
                 count++;
             }
 
             // Get the rest of all 5 records.
             records = waitedPoll(shareConsumer, 2500L, 5);
             assertEquals(5, records.count());
+            assertEquals(Optional.of(25000), shareConsumer.acquisitionLockTimeoutMs());
             for (ConsumerRecord<byte[], byte[]> record : records) {
                 shareConsumer.acknowledge(record, AcknowledgeType.ACCEPT);
             }
@@ -3441,6 +3447,19 @@ private void alterShareIsolationLevel(String groupId, String newValue) {
         }
     }
 
+    private void alterShareRecordLockDurationMs(String groupId, int newValue) {
+        ConfigResource configResource = new ConfigResource(ConfigResource.Type.GROUP, groupId);
+        Map<ConfigResource, Collection<AlterConfigOp>> alterEntries = new HashMap<>();
+        alterEntries.put(configResource, List.of(new AlterConfigOp(new ConfigEntry(
+            GroupConfig.SHARE_RECORD_LOCK_DURATION_MS_CONFIG, Integer.toString(newValue)), AlterConfigOp.OpType.SET)));
+        AlterConfigsOptions alterOptions = new AlterConfigsOptions();
+        try (Admin adminClient = createAdminClient()) {
+            assertDoesNotThrow(() -> adminClient.incrementalAlterConfigs(alterEntries, alterOptions)
+                .all()
+                .get(60, TimeUnit.SECONDS), "Failed to alter configs");
+        }
+    }
+
     /**
      * Test utility which encapsulates a {@link ShareConsumer} whose record processing
      * behavior can be supplied as a function argument.

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

@@ -4262,7 +4262,7 @@ public ListOffsetsResult listOffsets(Map<TopicPartition, OffsetSpec> topicPartit
             ListOffsetsHandler.newFuture(topicPartitionOffsets.keySet(), partitionLeaderCache);
         Map<TopicPartition, Long> offsetQueriesByPartition = topicPartitionOffsets.entrySet().stream()
             .collect(Collectors.toMap(Map.Entry::getKey, e -> getOffsetFromSpec(e.getValue())));
-        ListOffsetsHandler handler = new ListOffsetsHandler(offsetQueriesByPartition, options, logContext, defaultApiTimeoutMs);
+        ListOffsetsHandler handler = new ListOffsetsHandler(offsetQueriesByPartition, metadataManager.cluster(), options, logContext, defaultApiTimeoutMs);
         invokeDriver(handler, future, options.timeoutMs);
         return new ListOffsetsResult(future.all());
     }

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

@@ -318,6 +318,10 @@ public void update(Cluster cluster, long now) {
         }
     }
 
+    public Cluster cluster() {
+        return cluster;
+    }
+
     public void initiateRebootstrap() {
         this.metadataAttemptStartMs = Optional.of(0L);
     }

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

@@ -19,8 +19,10 @@
 import org.apache.kafka.clients.admin.ListOffsetsOptions;
 import org.apache.kafka.clients.admin.ListOffsetsResult.ListOffsetsResultInfo;
 import org.apache.kafka.clients.admin.internals.AdminApiHandler.Batched;
+import org.apache.kafka.common.Cluster;
 import org.apache.kafka.common.Node;
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.errors.ApiException;
 import org.apache.kafka.common.errors.RetriableException;
 import org.apache.kafka.common.errors.UnsupportedVersionException;
@@ -43,25 +45,29 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
 
 public final class ListOffsetsHandler extends Batched<TopicPartition, ListOffsetsResultInfo> {
 
     private final Map<TopicPartition, Long> offsetTimestampsByPartition;
+    private final Cluster cluster;
     private final ListOffsetsOptions options;
     private final Logger log;
     private final AdminApiLookupStrategy<TopicPartition> lookupStrategy;
     private final int defaultApiTimeoutMs;
 
     public ListOffsetsHandler(
         Map<TopicPartition, Long> offsetTimestampsByPartition,
+        Cluster cluster,
         ListOffsetsOptions options,
         LogContext logContext,
         int defaultApiTimeoutMs
     ) {
         this.offsetTimestampsByPartition = offsetTimestampsByPartition;
+        this.cluster = cluster;
         this.options = options;
         this.log = logContext.logger(ListOffsetsHandler.class);
         this.lookupStrategy = new PartitionLeaderStrategy(logContext, false);
@@ -82,7 +88,7 @@ public AdminApiLookupStrategy<TopicPartition> lookupStrategy() {
     ListOffsetsRequest.Builder buildBatchedRequest(int brokerId, Set<TopicPartition> keys) {
         Map<String, ListOffsetsTopic> topicsByName = CollectionUtils.groupPartitionsByTopic(
             keys,
-            topicName -> new ListOffsetsTopic().setName(topicName),
+            topicName -> new ListOffsetsTopic().setName(topicName).setTopicId(cluster.topicId(topicName)),
             (listOffsetsTopic, partitionId) -> {
                 TopicPartition topicPartition = new TopicPartition(listOffsetsTopic.name(), partitionId);
                 long offsetTimestamp = offsetTimestampsByPartition.get(topicPartition);
@@ -91,6 +97,12 @@ ListOffsetsRequest.Builder buildBatchedRequest(int brokerId, Set<TopicPartition>
                         .setPartitionIndex(partitionId)
                         .setTimestamp(offsetTimestamp));
             });
+
+        boolean supportsTopicIds = topicsByName.values().stream()
+                .filter(Objects::nonNull)
+                .map(ListOffsetsTopic::topicId)
+                .anyMatch(topicId -> topicId != null && !topicId.equals(Uuid.ZERO_UUID));
+
         boolean supportsMaxTimestamp = keys
             .stream()
             .anyMatch(key -> offsetTimestampsByPartition.get(key) == ListOffsetsRequest.MAX_TIMESTAMP);
@@ -110,6 +122,7 @@ ListOffsetsRequest.Builder buildBatchedRequest(int brokerId, Set<TopicPartition>
         int timeoutMs = options.timeoutMs() != null ? options.timeoutMs() : defaultApiTimeoutMs;
         return ListOffsetsRequest.Builder.forConsumer(true,
                         options.isolationLevel(),
+                        supportsTopicIds,
                         supportsMaxTimestamp,
                         requireEarliestLocalTimestamp,
                         requireTieredStorageTimestamp,