MINOR: Refactor JmxTool and StreamsResetter to use CommandLineUtils#maybePrintHelpOrVersion

.github/workflows/docker_scan.yml

@@ -26,7 +26,7 @@ jobs:
     strategy:
       matrix:
         # This is an array of supported tags. Make sure this array only contains the supported tags
-        supported_image_tag: ['latest', '3.7.2', '3.8.1', '3.9.1', '4.0.0']
+        supported_image_tag: ['latest', '3.9.1', '4.0.0', '4.1.0']
     steps:
       - name: Run CVE scan
         uses: aquasecurity/trivy-action@6e7b7d1fd3e4fef0c5fa8cce1229c54b2c9bd0d8 # v0.24.0

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

@@ -46,6 +46,7 @@
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
@@ -54,7 +55,6 @@
 import java.util.UUID;
 import java.util.concurrent.ExecutionException;
 import java.util.stream.Collectors;
-import java.util.stream.Stream;
 
 import static java.util.Arrays.asList;
 import static org.apache.kafka.clients.admin.AdminClientConfig.METRIC_REPORTER_CLASSES_CONFIG;
@@ -123,13 +123,12 @@ public void testClientInstanceId(ClusterInstance clusterInstance) throws Interru
         }
     }
 
-    @SuppressWarnings("unchecked")
     @ClusterTest(types = {Type.CO_KRAFT, Type.KRAFT})
     public void testIntervalMsParser(ClusterInstance clusterInstance) {
         List<String> alterOpts = asList("--bootstrap-server", clusterInstance.bootstrapServers(),
                 "--alter", "--entity-type", "client-metrics", "--entity-name", "test", "--add-config", "interval.ms=bbb");
         try (Admin client = clusterInstance.admin()) {
-            ConfigCommand.ConfigCommandOptions addOpts = new ConfigCommand.ConfigCommandOptions(toArray(alterOpts));
+            ConfigCommand.ConfigCommandOptions addOpts = new ConfigCommand.ConfigCommandOptions(toArray(Set.of(alterOpts)));
 
             Throwable e = assertThrows(ExecutionException.class, () -> ConfigCommand.alterConfig(client, addOpts));
             assertTrue(e.getMessage().contains(InvalidConfigurationException.class.getSimpleName()));
@@ -153,16 +152,16 @@ public void testMetrics(ClusterInstance clusterInstance) {
         }
     }
 
-    @SuppressWarnings("unchecked")
-    private static String[] toArray(List<String>... lists) {
-        return Stream.of(lists).flatMap(List::stream).toArray(String[]::new);
+    private static String[] toArray(Collection<List<String>> lists) {
+        return lists.stream().flatMap(List::stream).toArray(String[]::new);
     }
 
     /**
      * We should add a ClientTelemetry into plugins to test the clientInstanceId method Otherwise the
      * {@link org.apache.kafka.common.protocol.ApiKeys#GET_TELEMETRY_SUBSCRIPTIONS} command will not be supported
      * by the server
      **/
+    @SuppressWarnings("unused")
     public static class GetIdClientTelemetry implements ClientTelemetry, MetricsReporter {
 
 

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

@@ -0,0 +1,143 @@
+/*
+ * 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.Producer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.NotLeaderOrFollowerException;
+import org.apache.kafka.common.test.ClusterInstance;
+import org.apache.kafka.common.test.api.ClusterTest;
+import org.apache.kafka.common.test.api.ClusterTestDefaults;
+import org.apache.kafka.test.TestUtils;
+
+import org.junit.jupiter.api.BeforeEach;
+
+import java.util.List;
+
+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 = 3
+)
+class DescribeProducersWithBrokerIdTest {
+    private static final String TOPIC_NAME = "test-topic";
+    private static final int NUM_PARTITIONS = 1;
+    private static final short REPLICATION_FACTOR = 3;
+
+    private static final TopicPartition TOPIC_PARTITION = new TopicPartition(TOPIC_NAME, 0);
+
+    private final ClusterInstance clusterInstance;
+
+    public DescribeProducersWithBrokerIdTest(ClusterInstance clusterInstance) {
+        this.clusterInstance = clusterInstance;
+    }
+
+    private static void sendTestRecords(Producer<byte[], byte[]> producer) {
+        producer.send(new ProducerRecord<>(TOPIC_NAME, TOPIC_PARTITION.partition(), "key-0".getBytes(), "value-0".getBytes()));
+        producer.flush();
+    }
+
+    @BeforeEach
+    void setUp() throws InterruptedException {
+        clusterInstance.createTopic(TOPIC_NAME, NUM_PARTITIONS, REPLICATION_FACTOR);
+    }
+
+    private List<Integer> getReplicaBrokerIds(Admin admin) throws Exception {
+        var topicDescription = admin.describeTopics(List.of(TOPIC_PARTITION.topic())).allTopicNames().get().get(TOPIC_PARTITION.topic());
+        return topicDescription.partitions().get(TOPIC_PARTITION.partition()).replicas().stream()
+            .map(Node::id)
+            .toList();
+    }
+
+    private int getNonReplicaBrokerId(Admin admin) throws Exception {
+        var replicaBrokerIds = getReplicaBrokerIds(admin);
+        return clusterInstance.brokerIds().stream()
+            .filter(id -> !replicaBrokerIds.contains(id))
+            .findFirst()
+            .orElseThrow(() -> new IllegalStateException("No non-replica broker found"));
+    }
+
+    private int getFollowerBrokerId(Admin admin) throws Exception {
+        var replicaBrokerIds = getReplicaBrokerIds(admin);
+        var leaderBrokerId = clusterInstance.getLeaderBrokerId(TOPIC_PARTITION);
+        return replicaBrokerIds.stream()
+            .filter(id -> id != leaderBrokerId)
+            .findFirst()
+            .orElseThrow(() -> new IllegalStateException("No follower found for partition " + TOPIC_PARTITION));
+    }
+
+    @ClusterTest
+    void testDescribeProducersDefaultRoutesToLeader() throws Exception {
+        try (Producer<byte[], byte[]> producer = clusterInstance.producer();
+             var admin = clusterInstance.admin()) {
+            sendTestRecords(producer);
+
+            var stateWithExplicitLeader = admin.describeProducers(
+                List.of(TOPIC_PARTITION), 
+                new DescribeProducersOptions().brokerId(clusterInstance.getLeaderBrokerId(TOPIC_PARTITION))
+            ).partitionResult(TOPIC_PARTITION).get();
+
+            var stateWithDefaultRouting = admin.describeProducers(
+                List.of(TOPIC_PARTITION)
+            ).partitionResult(TOPIC_PARTITION).get();
+
+            assertNotNull(stateWithDefaultRouting);
+            assertFalse(stateWithDefaultRouting.activeProducers().isEmpty());
+            assertEquals(stateWithExplicitLeader.activeProducers(), stateWithDefaultRouting.activeProducers());
+        }
+    }
+
+    @ClusterTest
+    void testDescribeProducersFromFollower() throws Exception {
+        try (Producer<byte[], byte[]> producer = clusterInstance.producer();
+             var admin = clusterInstance.admin()) {
+            sendTestRecords(producer);
+
+            var followerState = admin.describeProducers(
+                List.of(TOPIC_PARTITION), 
+                new DescribeProducersOptions().brokerId(getFollowerBrokerId(admin))
+            ).partitionResult(TOPIC_PARTITION).get();
+
+            var leaderState = admin.describeProducers(
+                List.of(TOPIC_PARTITION)
+            ).partitionResult(TOPIC_PARTITION).get();
+
+            assertNotNull(followerState);
+            assertFalse(followerState.activeProducers().isEmpty());
+            assertEquals(leaderState.activeProducers(), followerState.activeProducers());
+        }
+    }
+
+    @ClusterTest(brokers = 4)
+    void testDescribeProducersWithInvalidBrokerId() throws Exception {
+        try (Producer<byte[], byte[]> producer = clusterInstance.producer();
+             var admin = clusterInstance.admin()) {
+            sendTestRecords(producer);
+
+            TestUtils.assertFutureThrows(NotLeaderOrFollowerException.class, 
+                admin.describeProducers(
+                    List.of(TOPIC_PARTITION), 
+                    new DescribeProducersOptions().brokerId(getNonReplicaBrokerId(admin))
+                ).partitionResult(TOPIC_PARTITION)); 
+        }
+    }
+}

core/src/main/scala/kafka/server/KafkaConfig.scala

@@ -513,6 +513,13 @@ class KafkaConfig private(doLog: Boolean, val props: util.Map[_, _])
     require(replicaFetchWaitMaxMs <= replicaLagTimeMaxMs, "replica.fetch.wait.max.ms should always be less than or equal to replica.lag.time.max.ms" +
       " to prevent frequent changes in ISR")
 
+    if (brokerHeartbeatIntervalMs * 2 > brokerSessionTimeoutMs) {
+      error(s"${KRaftConfigs.BROKER_HEARTBEAT_INTERVAL_MS_CONFIG} ($brokerHeartbeatIntervalMs ms) must be less than or equal to half of the ${KRaftConfigs.BROKER_SESSION_TIMEOUT_MS_CONFIG} ($brokerSessionTimeoutMs ms). " +
+        s"The ${KRaftConfigs.BROKER_SESSION_TIMEOUT_MS_CONFIG} is configured on controller. The ${KRaftConfigs.BROKER_HEARTBEAT_INTERVAL_MS_CONFIG} is configured on broker. " +
+        s"If a broker doesn't send heartbeat request within ${KRaftConfigs.BROKER_SESSION_TIMEOUT_MS_CONFIG}, it loses broker lease. " +
+        s"Please increase ${KRaftConfigs.BROKER_SESSION_TIMEOUT_MS_CONFIG} or decrease ${KRaftConfigs.BROKER_HEARTBEAT_INTERVAL_MS_CONFIG}.")
+    }
+
     val advertisedBrokerListenerNames = effectiveAdvertisedBrokerListeners.map(l => ListenerName.normalised(l.listener)).toSet
 
     // validate KRaft-related configs

core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala

@@ -29,6 +29,7 @@ import org.apache.kafka.common.network.ListenerName
 import org.apache.kafka.common.record.{CompressionType, Records}
 import org.apache.kafka.common.security.auth.SecurityProtocol
 import org.apache.kafka.common.config.internals.BrokerSecurityConfigs
+import org.apache.kafka.common.utils.LogCaptureAppender
 import org.apache.kafka.coordinator.group.ConsumerGroupMigrationPolicy
 import org.apache.kafka.coordinator.group.Group.GroupType
 import org.apache.kafka.coordinator.group.GroupCoordinatorConfig
@@ -40,11 +41,13 @@ import org.apache.kafka.server.config.{DelegationTokenManagerConfigs, KRaftConfi
 import org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig
 import org.apache.kafka.server.metrics.MetricConfigs
 import org.apache.kafka.storage.internals.log.CleanerConfig
+import org.apache.logging.log4j.Level
 import org.junit.jupiter.api.Assertions._
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.function.Executable
 
 import scala.jdk.CollectionConverters._
+import scala.util.Using
 
 class KafkaConfigTest {
 
@@ -1899,4 +1902,19 @@ class KafkaConfigTest {
     val message = assertThrows(classOf[IllegalArgumentException], () => KafkaConfig.fromProps(props)).getMessage
     assertEquals("requirement failed: controller.listener.names must contain at least one value appearing in the 'listeners' configuration when running the KRaft controller role", message)
   }
+
+  @Test
+  def testLogBrokerHeartbeatIntervalMsShouldBeLowerThanHalfOfBrokerSessionTimeoutMs(): Unit = {
+    val props = createDefaultConfig()
+    Using.resource(LogCaptureAppender.createAndRegister) { appender =>
+      appender.setClassLogger(KafkaConfig.getClass, Level.ERROR)
+      props.setProperty(KRaftConfigs.BROKER_HEARTBEAT_INTERVAL_MS_CONFIG, "4500")
+      props.setProperty(KRaftConfigs.BROKER_SESSION_TIMEOUT_MS_CONFIG, "8999")
+      KafkaConfig.fromProps(props)
+      assertTrue(appender.getMessages.contains("broker.heartbeat.interval.ms (4500 ms) must be less than or equal to half of the broker.session.timeout.ms (8999 ms). " +
+        "The broker.session.timeout.ms is configured on controller. The broker.heartbeat.interval.ms is configured on broker. " +
+        "If a broker doesn't send heartbeat request within broker.session.timeout.ms, it loses broker lease. " +
+        "Please increase broker.session.timeout.ms or decrease broker.heartbeat.interval.ms."))
+    }
+  }
 }

server-common/src/test/java/org/apache/kafka/server/util/CommandLineUtilsTest.java

@@ -83,7 +83,7 @@ public void testParseArgsWithMultipleDelimiters() {
 
         assertEquals("=first", props.getProperty("first.property"), "Value of first property should be '=first'");
         assertEquals("second=", props.getProperty("second.property"), "Value of second property should be 'second='");
-        assertEquals("thi=rd", props.getProperty("third.property"), "Value of second property should be 'thi=rd'");
+        assertEquals("thi=rd", props.getProperty("third.property"), "Value of third property should be 'thi=rd'");
     }
 
     Properties props = new Properties();