Merge remote-tracking branch 'origin/trunk' into KAFKA-19019

Author: adixitconfluent

.github/scripts/pr-format.py

@@ -162,7 +162,10 @@ def check(positive_assertion, ok_msg, err_msg):
                 new_lines.append(textwrap.fill(line, width=72, break_long_words=False, break_on_hyphens=False, replace_whitespace=False))
             rewrapped_p = "\n".join(new_lines)
         else:
-            rewrapped_p = textwrap.fill("".join(p), width=72, break_long_words=False, break_on_hyphens=False, replace_whitespace=True)
+            indent = ""
+            if len(p) > 0 and p[0].startswith("Reviewers:"):
+                indent = " "
+            rewrapped_p = textwrap.fill("".join(p), subsequent_indent=indent, width=72, break_long_words=False, break_on_hyphens=False, replace_whitespace=True)
         new_paragraphs.append(rewrapped_p + "\n")
     body = "\n".join(new_paragraphs)
 

README.md

@@ -109,7 +109,7 @@ Using compiled files:
 
 Using docker image:
 
-    docker run -p 9092:9092 apache/kafka:3.7.0
+    docker run -p 9092:9092 apache/kafka:latest
 
 ### Cleaning the build ###
     ./gradlew clean

bin/kafka-streams-groups.sh

@@ -0,0 +1,17 @@
+#!/bin/bash
+# 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.
+
+exec $(dirname $0)/kafka-run-class.sh org.apache.kafka.tools.streams.StreamsGroupCommand "$@"

checkstyle/suppressions.xml

@@ -105,7 +105,7 @@
               files="(AbstractRequest|AbstractResponse|KerberosLogin|WorkerSinkTaskTest|TransactionManagerTest|SenderTest|KafkaAdminClient|ConsumerCoordinatorTest|KafkaAdminClientTest).java"/>
 
     <suppress checks="NPathComplexity"
-              files="(AbstractMembershipManager|ConsumerCoordinator|BufferPool|MetricName|Node|ConfigDef|RecordBatch|SslFactory|SslTransportLayer|MetadataResponse|KerberosLogin|Selector|Sender|Serdes|TokenInformation|Agent|PluginUtils|MiniTrogdorCluster|TasksRequest|KafkaProducer|AbstractStickyAssignor|Authorizer|FetchSessionHandler|RecordAccumulator|Shell).java"/>
+              files="(AbstractMembershipManager|ConsumerCoordinator|BufferPool|MetricName|Node|ConfigDef|RecordBatch|SslFactory|SslTransportLayer|MetadataResponse|KerberosLogin|Selector|Sender|Serdes|TokenInformation|Agent|PluginUtils|MiniTrogdorCluster|TasksRequest|KafkaProducer|AbstractStickyAssignor|Authorizer|FetchSessionHandler|RecordAccumulator|Shell|MockConsumer).java"/>
 
     <suppress checks="(JavaNCSS|CyclomaticComplexity|MethodLength)"
               files="CoordinatorClient.java"/>
@@ -168,6 +168,9 @@
     <suppress checks="NPathComplexity"
               files="(DistributedHerder|AbstractHerder|RestClient|RestServer|JsonConverter|KafkaConfigBackingStore|FileStreamSourceTask|WorkerSourceTask|TopicAdmin).java"/>
 
+    <suppress checks="ClassFanOutComplexity"
+              files="ShareConsumerTest.java"/>
+
     <!-- connect tests-->
     <suppress checks="ClassDataAbstractionCoupling"
               files="(DistributedHerder|KafkaBasedLog|WorkerSourceTaskWithTopicCreation|WorkerSourceTask)Test.java"/>

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

@@ -19,39 +19,44 @@
 import org.apache.kafka.clients.admin.Admin;
 import org.apache.kafka.clients.admin.NewTopic;
 import org.apache.kafka.common.Cluster;
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.metrics.Gauge;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Monitorable;
+import org.apache.kafka.common.metrics.PluginMetrics;
 import org.apache.kafka.common.security.auth.KafkaPrincipal;
 import org.apache.kafka.common.test.ClusterInstance;
 import org.apache.kafka.common.test.TestUtils;
 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.common.test.api.Type;
 import org.apache.kafka.server.config.QuotaConfig;
 
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicInteger;
 
-@ClusterTestDefaults(controllers = 3, 
-    types = {Type.KRAFT},
-    serverProperties = {
-        @ClusterConfigProperty(id = 3000, key = QuotaConfig.CLIENT_QUOTA_CALLBACK_CLASS_CONFIG, value = "org.apache.kafka.server.quota.CustomQuotaCallbackTest$CustomQuotaCallback"),
-        @ClusterConfigProperty(id = 3001, key = QuotaConfig.CLIENT_QUOTA_CALLBACK_CLASS_CONFIG, value = "org.apache.kafka.server.quota.CustomQuotaCallbackTest$CustomQuotaCallback"),
-        @ClusterConfigProperty(id = 3002, key = QuotaConfig.CLIENT_QUOTA_CALLBACK_CLASS_CONFIG, value = "org.apache.kafka.server.quota.CustomQuotaCallbackTest$CustomQuotaCallback"),
-    }
-)
-public class CustomQuotaCallbackTest {
+import static org.junit.jupiter.api.Assertions.assertEquals;
 
-    private final ClusterInstance cluster;
+public class CustomQuotaCallbackTest {
 
-    public CustomQuotaCallbackTest(ClusterInstance clusterInstance) {
-        this.cluster = clusterInstance;
+    private static int controllerId(Type type) {
+        return type == Type.KRAFT ? 3000 : 0;
     }
 
-    @ClusterTest
-    public void testCustomQuotaCallbackWithControllerServer() throws InterruptedException {
-        
+    @ClusterTest(
+        controllers = 3,
+        types = {Type.KRAFT},
+        serverProperties = {
+            @ClusterConfigProperty(id = 3000, key = QuotaConfig.CLIENT_QUOTA_CALLBACK_CLASS_CONFIG, value = "org.apache.kafka.server.quota.CustomQuotaCallbackTest$CustomQuotaCallback"),
+            @ClusterConfigProperty(id = 3001, key = QuotaConfig.CLIENT_QUOTA_CALLBACK_CLASS_CONFIG, value = "org.apache.kafka.server.quota.CustomQuotaCallbackTest$CustomQuotaCallback"),
+            @ClusterConfigProperty(id = 3002, key = QuotaConfig.CLIENT_QUOTA_CALLBACK_CLASS_CONFIG, value = "org.apache.kafka.server.quota.CustomQuotaCallbackTest$CustomQuotaCallback"),
+        }
+    )
+    public void testCustomQuotaCallbackWithControllerServer(ClusterInstance cluster) throws InterruptedException {
+
         try (Admin admin = cluster.admin(Map.of())) {
             admin.createTopics(List.of(new NewTopic("topic", 1, (short) 1)));
             TestUtils.waitForCondition(
@@ -69,10 +74,49 @@ public void testCustomQuotaCallbackWithControllerServer() throws InterruptedExce
                         && CustomQuotaCallback.COUNTERS.values().stream().allMatch(counter -> counter.get() > 0), 
                     "The CustomQuotaCallback not triggered in all controllers. "
             );
-            
+        
+        }
+    }
+
+    @ClusterTest(
+        types = {Type.CO_KRAFT, Type.KRAFT},
+        serverProperties = {
+            @ClusterConfigProperty(key = QuotaConfig.CLIENT_QUOTA_CALLBACK_CLASS_CONFIG, value = "org.apache.kafka.server.quota.CustomQuotaCallbackTest$MonitorableCustomQuotaCallback"),
         }
+    )
+    public void testMonitorableCustomQuotaCallbackWithCombinedMode(ClusterInstance cluster) {
+        assertMetrics(
+            cluster.brokers().get(0).metrics(),
+            expectedTags(Map.of("role", "broker"))
+        );
+        assertMetrics(
+            cluster.controllers().get(controllerId(cluster.type())).metrics(),
+            expectedTags(Map.of("role", "controller"))
+        );
     }
 
+    private void assertMetrics(Metrics metrics, Map<String, String> expectedTags) {
+        int found = 0;
+        for (MetricName metricName : metrics.metrics().keySet()) {
+            if (metricName.group().equals("plugins")) {
+                Map<String, String> tags = metricName.tags();
+                if (expectedTags.equals(tags)) {
+                    assertEquals(MonitorableCustomQuotaCallback.METRIC_NAME, metricName.name());
+                    assertEquals(MonitorableCustomQuotaCallback.METRIC_DESCRIPTION, metricName.description());
+                    found++;
+                }
+            }
+        }
+        assertEquals(1, found);
+    }
+
+    private static Map<String, String> expectedTags(Map<String, String> extraTags) {
+        Map<String, String> tags = new LinkedHashMap<>();
+        tags.put("config", QuotaConfig.CLIENT_QUOTA_CALLBACK_CLASS_CONFIG);
+        tags.put("class", MonitorableCustomQuotaCallback.class.getSimpleName());
+        tags.putAll(extraTags);
+        return tags;
+    }
 
     public static class CustomQuotaCallback implements ClientQuotaCallback {
 
@@ -121,4 +165,17 @@ public void configure(Map<String, ?> configs) {
         }
 
     }
+
+    public static class MonitorableCustomQuotaCallback extends CustomQuotaCallback implements Monitorable {
+        
+        private static final String METRIC_NAME = "monitorable-custom-quota-callback-name";
+        private static final String METRIC_DESCRIPTION = "monitorable-custom-quota-callback-description";
+
+        @Override
+        public void withPluginMetrics(PluginMetrics metrics) {
+            MetricName metricName = metrics.metricName(METRIC_NAME, METRIC_DESCRIPTION, Map.of());
+            metrics.addMetric(metricName, (Gauge<Integer>) (config, now) -> 1);
+        }
+        
+    }
 }

clients/clients-integration-tests/src/test/java/org/apache/kafka/server/quota/CustomQuotaCallbackTest.java

@@ -1947,13 +1947,28 @@ default ListShareGroupOffsetsResult listShareGroupOffsets(Map<String, ListShareG
     }
 
     /**
-     * Delete share groups from the cluster with the default options.
+     * Delete offsets for a set of partitions in a share group.
      *
-     * @param groupIds Collection of share group ids which are to be deleted.
-     * @return The DeleteShareGroupsResult.
+     * @param groupId The group for which to delete offsets.
+     * @param partitions The topic-partitions.
+     * @param options The options to use when deleting offsets in a share group.
+     * @return The DeleteShareGroupOffsetsResult.
      */
-    default DeleteShareGroupsResult deleteShareGroups(Collection<String> groupIds) {
-        return deleteShareGroups(groupIds, new DeleteShareGroupsOptions());
+    DeleteShareGroupOffsetsResult deleteShareGroupOffsets(String groupId, Set<TopicPartition> partitions, DeleteShareGroupOffsetsOptions options);
+
+    /**
+     * Delete offsets for a set of partitions in a share group with the default options.
+     *
+     * <p>
+     * This is a convenience method for {@link #deleteShareGroupOffsets(String, Set, DeleteShareGroupOffsetsOptions)} with default options.
+     * See the overload for more details.
+     *
+     * @param groupId The group for which to delete offsets.
+     * @param partitions The topic-partitions.
+     * @return The DeleteShareGroupOffsetsResult.
+     */
+    default DeleteShareGroupOffsetsResult deleteShareGroupOffsets(String groupId, Set<TopicPartition> partitions) {
+        return deleteShareGroupOffsets(groupId, partitions, new DeleteShareGroupOffsetsOptions());
     }
 
     /**
@@ -1965,6 +1980,16 @@ default DeleteShareGroupsResult deleteShareGroups(Collection<String> groupIds) {
      */
     DeleteShareGroupsResult deleteShareGroups(Collection<String> groupIds, DeleteShareGroupsOptions options);
 
+    /**
+     * Delete share groups from the cluster with the default options.
+     *
+     * @param groupIds Collection of share group ids which are to be deleted.
+     * @return The DeleteShareGroupsResult.
+     */
+    default DeleteShareGroupsResult deleteShareGroups(Collection<String> groupIds) {
+        return deleteShareGroups(groupIds, new DeleteShareGroupsOptions());
+    }
+
     /**
      * Describe streams groups in the cluster.
      *

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

@@ -14,36 +14,18 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.kafka.server.purgatory;
 
-import java.util.Objects;
+package org.apache.kafka.clients.admin;
 
-/**
- * Used by delayed-sync operations
- */
-public class GroupSyncKey implements DelayedOperationKey {
-
-    private final String groupId;
-
-    public GroupSyncKey(String groupId) {
-        this.groupId = groupId;
-    }
+import org.apache.kafka.common.annotation.InterfaceStability;
 
-    @Override
-    public String keyLabel() {
-        return "sync-" + groupId;
-    }
+import java.util.Set;
 
-    @Override
-    public boolean equals(Object o) {
-        if (this == o) return true;
-        if (o == null || getClass() != o.getClass()) return false;
-        GroupSyncKey that = (GroupSyncKey) o;
-        return Objects.equals(groupId, that.groupId);
-    }
-
-    @Override
-    public int hashCode() {
-        return Objects.hash(groupId);
-    }
+/**
+ * Options for the {@link Admin#deleteShareGroupOffsets(String, Set, DeleteShareGroupOffsetsOptions)} call.
+ * <p>
+ * The API of this class is evolving, see {@link Admin} for details.
+ */
+@InterfaceStability.Evolving
+public class DeleteShareGroupOffsetsOptions extends AbstractOptions<DeleteShareGroupOffsetsOptions> {
 }

…kafka/server/purgatory/GroupSyncKey.java …dmin/DeleteShareGroupOffsetsOptions.javaserver-common/src/main/java/org/apache/kafka/server/purgatory/GroupSyncKey.java renamed to clients/src/main/java/org/apache/kafka/clients/admin/DeleteShareGroupOffsetsOptions.java server-common/src/main/java/org/apache/kafka/server/purgatory/GroupSyncKey.java renamed to clients/src/main/java/org/apache/kafka/clients/admin/DeleteShareGroupOffsetsOptions.java

@@ -0,0 +1,102 @@
+/*
+ * 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.common.KafkaFuture;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.annotation.InterfaceStability;
+import org.apache.kafka.common.errors.ApiException;
+import org.apache.kafka.common.internals.KafkaFutureImpl;
+
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * The result of the {@link Admin#deleteShareGroupOffsets(String, Set, DeleteShareGroupOffsetsOptions)} call.
+ * <p>
+ * The API of this class is evolving, see {@link Admin} for details.
+ */
+@InterfaceStability.Evolving
+public class DeleteShareGroupOffsetsResult {
+
+    private final KafkaFuture<Map<TopicPartition, ApiException>> future;
+    private final Set<TopicPartition> partitions;
+
+    DeleteShareGroupOffsetsResult(KafkaFuture<Map<TopicPartition, ApiException>> future, Set<TopicPartition> partitions) {
+        this.future = future;
+        this.partitions = partitions;
+    }
+
+    /**
+     * Return a future which succeeds only if all the deletions succeed.
+     * If not, the first partition error shall be returned.
+     */
+    public KafkaFuture<Void> all() {
+        final KafkaFutureImpl<Void> result = new KafkaFutureImpl<>();
+
+        this.future.whenComplete((topicPartitions, throwable) -> {
+            if (throwable != null) {
+                result.completeExceptionally(throwable);
+            } else {
+                for (TopicPartition partition : partitions) {
+                    if (maybeCompleteExceptionally(topicPartitions, partition, result)) {
+                        return;
+                    }
+                }
+                result.complete(null);
+            }
+        });
+        return result;
+    }
+
+    /**
+     * Return a future which can be used to check the result for a given partition.
+     */
+    public KafkaFuture<Void> partitionResult(final TopicPartition partition) {
+        if (!partitions.contains(partition)) {
+            throw new IllegalArgumentException("Partition " + partition + " was not included in the original request");
+        }
+        final KafkaFutureImpl<Void> result = new KafkaFutureImpl<>();
+
+        this.future.whenComplete((topicPartitions, throwable) -> {
+            if (throwable != null) {
+                result.completeExceptionally(throwable);
+            } else if (!maybeCompleteExceptionally(topicPartitions, partition, result)) {
+                result.complete(null);
+            }
+        });
+        return result;
+    }
+
+    private boolean maybeCompleteExceptionally(Map<TopicPartition, ApiException> partitionLevelErrors,
+                                               TopicPartition partition,
+                                               KafkaFutureImpl<Void> result) {
+        Throwable exception;
+        if (!partitionLevelErrors.containsKey(partition)) {
+            exception = new IllegalArgumentException("Offset deletion result for partition \"" + partition + "\" was not included in the response");
+        } else {
+            exception = partitionLevelErrors.get(partition);
+        }
+
+        if (exception != null) {
+            result.completeExceptionally(exception);
+            return true;
+        } else {
+            return false;
+        }
+    }
+}