Merge branch 'apache:trunk' into followupRemoteStorage

Author: jiafu1115

.github/actions/setup-gradle/action.yml

@@ -42,7 +42,7 @@ runs:
         distribution: temurin
         java-version: ${{ inputs.java-version }}
     - name: Setup Gradle
-      uses: gradle/actions/setup-gradle@4d9f0ba0025fe599b4ebab900eb7f3a1d93ef4c2 # v5.0.0
+      uses: gradle/actions/setup-gradle@0723195856401067f7a2779048b490ace7a47d7c # v5.0.2
       env:
         GRADLE_BUILD_ACTION_CACHE_DEBUG_ENABLED: true
       with:

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

@@ -249,7 +249,7 @@ class DynamicBrokerConfigTest {
 
     val props3 = new Properties()
     props3.put(RemoteLogManagerConfig.REMOTE_LOG_READER_THREADS_PROP, "-1")
-    val err3 = assertThrows(classOf[ConfigException], () => config.dynamicConfig.validate(props, perBrokerConfig = true))
+    val err3 = assertThrows(classOf[ConfigException], () => config.dynamicConfig.validate(props3, perBrokerConfig = true))
     assertTrue(err3.getMessage.contains("Value must be at least 1"))
     verifyNoMoreInteractions(remoteLogManager)
 
@@ -771,35 +771,35 @@ class DynamicBrokerConfigTest {
   }
 
   @Test
-  def testDynamicRemoteCopyLagThrowsOnIncorrectConfig(): Unit = {
-    // remote copy lag ms cannot exceed effective local retention ms
-    verifyIncorrectRemoteCopyLagProps(
+  def testDynamicLogRemoteCopyLagThrowsOnIncorrectConfig(): Unit = {
+    // log remote copy lag ms cannot exceed effective log local retention ms
+    verifyIncorrectLogRemoteCopyLagProps(
       retentionMs = 1000L,
       logLocalRetentionMs = -2L,
-      remoteCopyLagMs = 1001L,
+      logRemoteCopyLagMs = 1001L,
       retentionBytes = 1000L,
       logLocalRetentionBytes = -2L,
-      remoteCopyLagBytes = 100L
+      logRemoteCopyLagBytes = 100L
     )
 
-    // remote copy lag bytes cannot exceed effective local retention bytes
-    verifyIncorrectRemoteCopyLagProps(
+    // log remote copy lag bytes cannot exceed effective log local retention bytes
+    verifyIncorrectLogRemoteCopyLagProps(
       retentionMs = 1000L,
       logLocalRetentionMs = -2L,
-      remoteCopyLagMs = 100L,
+      logRemoteCopyLagMs = 100L,
       retentionBytes = 1000L,
       logLocalRetentionBytes = -2L,
-      remoteCopyLagBytes = 1001L
+      logRemoteCopyLagBytes = 1001L
     )
 
   }
 
-  def verifyIncorrectRemoteCopyLagProps(retentionMs: Long,
-                                        logLocalRetentionMs: Long,
-                                        remoteCopyLagMs: Long,
-                                        retentionBytes: Long,
-                                        logLocalRetentionBytes: Long,
-                                        remoteCopyLagBytes: Long): Unit = {
+  def verifyIncorrectLogRemoteCopyLagProps(retentionMs: Long,
+                                           logLocalRetentionMs: Long,
+                                           logRemoteCopyLagMs: Long,
+                                           retentionBytes: Long,
+                                           logLocalRetentionBytes: Long,
+                                           logRemoteCopyLagBytes: Long): Unit = {
     val props = TestUtils.createBrokerConfig(0, port = 8181)
     props.put(ServerLogConfigs.LOG_RETENTION_TIME_MILLIS_CONFIG, retentionMs.toString)
     props.put(ServerLogConfigs.LOG_RETENTION_BYTES_CONFIG, retentionBytes.toString)
@@ -810,9 +810,9 @@ class DynamicBrokerConfigTest {
 
     val newProps = new Properties()
     newProps.put(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_MS_PROP, logLocalRetentionMs.toString)
-    newProps.put(RemoteLogManagerConfig.LOG_REMOTE_COPY_LAG_MS_PROP, remoteCopyLagMs.toString)
+    newProps.put(RemoteLogManagerConfig.LOG_REMOTE_COPY_LAG_MS_PROP, logRemoteCopyLagMs.toString)
     newProps.put(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_BYTES_PROP, logLocalRetentionBytes.toString)
-    newProps.put(RemoteLogManagerConfig.LOG_REMOTE_COPY_LAG_BYTES_PROP, remoteCopyLagBytes.toString)
+    newProps.put(RemoteLogManagerConfig.LOG_REMOTE_COPY_LAG_BYTES_PROP, logRemoteCopyLagBytes.toString)
     // validate default config
     assertThrows(classOf[ConfigException], () => config.dynamicConfig.validate(newProps, perBrokerConfig = false))
     // validate per broker config

server/src/main/java/org/apache/kafka/server/config/AbstractKafkaConfig.java

@@ -349,20 +349,24 @@ private static String parseListenerName(String connectionString) {
      * <p>
      * This method exists to support migration from kafka.server.KafkaConfig (Scala/core) to AbstractKafkaConfig (Java/server).
      * When migrating code, replace KafkaConfig references with AbstractKafkaConfig.
+     * Subclasses should override this method to integrate with their dynamic configuration mechanism;
+     * the default implementation is a no-op.
      *
      * @param reconfigurable the component to register for configuration updates
      */
-    public abstract void addReconfigurable(Reconfigurable reconfigurable);
+    public void addReconfigurable(Reconfigurable reconfigurable) { }
 
     /**
      * Unregisters a component from dynamic reconfiguration notifications.
      * <p>
      * This method exists to support migration from kafka.server.KafkaConfig (Scala/core) to AbstractKafkaConfig (Java/server).
      * When migrating code, replace KafkaConfig references with AbstractKafkaConfig.
+     * Subclasses should override this method to integrate with their dynamic configuration mechanism;
+     * the default implementation is a no-op.
      *
      * @param reconfigurable the component to unregister
      */
-    public abstract void removeReconfigurable(Reconfigurable reconfigurable);
+    public void removeReconfigurable(Reconfigurable reconfigurable) { }
 
     /**
      * Determines whether a config entry might be sensitive based on its type.

server/src/test/java/org/apache/kafka/server/DefaultAutoTopicCreationManagerTest.java

@@ -100,14 +100,7 @@ public void setup() {
         props.setProperty(ServerLogConfigs.NUM_PARTITIONS_CONFIG, "1");
         props.setProperty(ReplicationConfigs.DEFAULT_REPLICATION_FACTOR_CONFIG, "1");
 
-        config = new AbstractKafkaConfig(AbstractKafkaConfig.CONFIG_DEF, props, Map.of(), false) {
-            @Override
-            public void addReconfigurable(org.apache.kafka.common.Reconfigurable reconfigurable) {
-            }
-            @Override
-            public void removeReconfigurable(org.apache.kafka.common.Reconfigurable reconfigurable) {
-            }
-        };
+        config = new AbstractKafkaConfig(AbstractKafkaConfig.CONFIG_DEF, props, Map.of(), false) { };
         topicCreator.reset();
     }
 

server/src/test/java/org/apache/kafka/server/config/AbstractKafkaConfigTest.java

@@ -16,7 +16,6 @@
  */
 package org.apache.kafka.server.config;
 
-import org.apache.kafka.common.Reconfigurable;
 import org.apache.kafka.common.config.ConfigDef;
 import org.apache.kafka.coordinator.group.GroupConfig;
 import org.apache.kafka.raft.KRaftConfigs;
@@ -101,13 +100,7 @@ private static Map<String, Object> extractGroupConfigMap(Map<String, Object> bro
             ConfigDef configDef = new ConfigDef().define(TEST_INTERNAL_GROUP_CONFIG_BROKER_SYNONYM, ConfigDef.Type.STRING,
                 "default-value", ConfigDef.Importance.LOW, "test broker synonym");
 
-            AbstractKafkaConfig kafkaConfig = new AbstractKafkaConfig(configDef, new HashMap<>(brokerProps), Map.of(), false) {
-                @Override
-                public void addReconfigurable(Reconfigurable reconfigurable) { }
-
-                @Override
-                public void removeReconfigurable(Reconfigurable reconfigurable) { }
-            };
+            AbstractKafkaConfig kafkaConfig = new AbstractKafkaConfig(configDef, new HashMap<>(brokerProps), Map.of(), false) { };
 
             return kafkaConfig.extractGroupConfigMap();
         }

server/src/test/java/org/apache/kafka/server/transaction/AddPartitionsToTxnManagerTest.java

@@ -90,17 +90,7 @@ public class AddPartitionsToTxnManagerTest {
                 KRaftConfigs.NODE_ID_CONFIG, "1",
                 KRaftConfigs.CONTROLLER_LISTENER_NAMES_CONFIG, "CONTROLLER"),
             Map.of(),
-            false) {
-        @Override
-        public void addReconfigurable(org.apache.kafka.common.Reconfigurable reconfigurable) {
-            // No-op for test
-        }
-
-        @Override
-        public void removeReconfigurable(org.apache.kafka.common.Reconfigurable reconfigurable) {
-            // No-op for test
-        }
-    };
+            false) { };
     private final AddPartitionsToTxnManager addPartitionsToTxnManager =
             new AddPartitionsToTxnManager(config, networkClient, metadataCache, partitionFor, time);