You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ka...@apache.org on 2025/12/12 17:56:20 UTC
(kafka) branch trunk updated: KAFKA-17266: Add dynamic broker config to enable follower fetch using tiered offset (#16834)
This is an automated email from the ASF dual-hosted git repository.
kamalcph pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 25a8ae0713c KAFKA-17266: Add dynamic broker config to enable follower fetch using tiered offset (#16834)
25a8ae0713c is described below
commit 25a8ae0713c19722422ad07bb9af3c0538f18bb9
Author: Abhijeet Kumar <ab...@gmail.com>
AuthorDate: Fri Dec 12 23:26:12 2025 +0530
KAFKA-17266: Add dynamic broker config to enable follower fetch using tiered offset (#16834)
In this PR, we add a dynamic broker config to enable/disable follower
fetch using tiered offset. This is part of the implementation for
[KIP-1023](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1023%3A+Follower+fetch+from+tiered+offset)
Added tests for the new dynamic broker config.
Reviewers: Kamal Chandraprakash <ka...@gmail.com>
---
.../scala/kafka/server/DynamicBrokerConfig.scala | 24 ++++++++++++++++++++-
core/src/main/scala/kafka/server/KafkaConfig.scala | 1 +
.../kafka/server/DynamicBrokerConfigTest.scala | 25 ++++++++++++++++++++++
.../kafka/server/config/ReplicationConfigs.java | 7 +++++-
4 files changed, 55 insertions(+), 2 deletions(-)
diff --git a/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala b/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala
index 51d7dbe5416..d580eda7697 100755
--- a/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala
+++ b/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala
@@ -44,7 +44,7 @@ import org.apache.kafka.network.SocketServerConfigs
import org.apache.kafka.raft.KafkaRaftClient
import org.apache.kafka.server.{DynamicThreadPool, ProcessRole}
import org.apache.kafka.server.common.ApiMessageAndVersion
-import org.apache.kafka.server.config.{DynamicProducerStateManagerConfig, ServerConfigs, ServerLogConfigs, ServerTopicConfigSynonyms}
+import org.apache.kafka.server.config.{DynamicProducerStateManagerConfig, ReplicationConfigs, ServerConfigs, ServerLogConfigs, ServerTopicConfigSynonyms}
import org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig
import org.apache.kafka.server.metrics.{ClientTelemetryExporterPlugin, MetricConfigs}
import org.apache.kafka.server.telemetry.{ClientTelemetry, ClientTelemetryExporterProvider}
@@ -101,6 +101,7 @@ object DynamicBrokerConfig {
SocketServer.ReconfigurableConfigs ++
DynamicProducerStateManagerConfig ++
DynamicRemoteLogConfig.ReconfigurableConfigs ++
+ DynamicReplicationConfig.ReconfigurableConfigs ++
Set(AbstractConfig.CONFIG_PROVIDERS_CONFIG) ++
GroupCoordinatorConfig.RECONFIGURABLE_CONFIGS.asScala ++
ShareCoordinatorConfig.RECONFIGURABLE_CONFIGS.asScala
@@ -311,6 +312,7 @@ class DynamicBrokerConfig(private val kafkaConfig: KafkaConfig) extends Logging
addBrokerReconfigurable(kafkaServer.socketServer)
addBrokerReconfigurable(new DynamicProducerStateManagerConfig(kafkaServer.logManager.producerStateManagerConfig))
addBrokerReconfigurable(new DynamicRemoteLogConfig(kafkaServer))
+ addBrokerReconfigurable(new DynamicReplicationConfig(kafkaServer))
}
/**
@@ -1134,3 +1136,23 @@ object DynamicRemoteLogConfig {
RemoteLogManagerConfig.REMOTE_LOG_READER_THREADS_PROP
)
}
+
+class DynamicReplicationConfig(server: KafkaBroker) extends BrokerReconfigurable with Logging {
+ override def reconfigurableConfigs: Set[String] = {
+ DynamicReplicationConfig.ReconfigurableConfigs
+ }
+
+ override def validateReconfiguration(newConfig: KafkaConfig): Unit = {
+ // Currently it is a noop for reconfiguring the dynamic config follower.fetch.last.tiered.offset.enable
+ }
+
+ override def reconfigure(oldConfig: KafkaConfig, newConfig: KafkaConfig): Unit = {
+ // Currently it is a noop for reconfiguring the dynamic config follower.fetch.last.tiered.offset.enable
+ }
+}
+
+object DynamicReplicationConfig {
+ val ReconfigurableConfigs = Set(
+ ReplicationConfigs.FOLLOWER_FETCH_LAST_TIERED_OFFSET_ENABLE_CONFIG
+ )
+}
diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala
index a7317a20e89..a96183e1981 100755
--- a/core/src/main/scala/kafka/server/KafkaConfig.scala
+++ b/core/src/main/scala/kafka/server/KafkaConfig.scala
@@ -356,6 +356,7 @@ class KafkaConfig private(doLog: Boolean, val props: util.Map[_, _])
val leaderImbalanceCheckIntervalSeconds: Long = getLong(ReplicationConfigs.LEADER_IMBALANCE_CHECK_INTERVAL_SECONDS_CONFIG)
val uncleanLeaderElectionCheckIntervalMs: Long = getLong(ReplicationConfigs.UNCLEAN_LEADER_ELECTION_INTERVAL_MS_CONFIG)
def uncleanLeaderElectionEnable: java.lang.Boolean = getBoolean(ReplicationConfigs.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG)
+ def followerFetchLastTieredOffsetEnable: java.lang.Boolean = getBoolean(ReplicationConfigs.FOLLOWER_FETCH_LAST_TIERED_OFFSET_ENABLE_CONFIG);
/** ********* Controlled shutdown configuration ***********/
val controlledShutdownEnable = getBoolean(ServerConfigs.CONTROLLED_SHUTDOWN_ENABLE_CONFIG)
diff --git a/core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala b/core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala
index 8504f3705ac..8c162476461 100755
--- a/core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala
+++ b/core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala
@@ -988,6 +988,31 @@ class DynamicBrokerConfigTest {
verifyNoMoreInteractions(remoteLogManager)
}
+ @Test
+ def testEnableFollowerFetchLastTieredOffset(): Unit = {
+ val props = TestUtils.createBrokerConfig(0, port = 9092)
+ val config = KafkaConfig.fromProps(props)
+ val serverMock: KafkaBroker = mock(classOf[KafkaBroker])
+
+ Mockito.when(serverMock.config).thenReturn(config)
+
+ config.dynamicConfig.initialize(None)
+ config.dynamicConfig.addBrokerReconfigurable(new DynamicReplicationConfig(serverMock))
+
+ assertEquals(ReplicationConfigs.FOLLOWER_FETCH_LAST_TIERED_OFFSET_ENABLE_DEFAULT,
+ config.followerFetchLastTieredOffsetEnable)
+
+ // Update default config
+ props.put(ReplicationConfigs.FOLLOWER_FETCH_LAST_TIERED_OFFSET_ENABLE_CONFIG, "true")
+ config.dynamicConfig.updateDefaultConfig(props)
+ assertTrue(config.followerFetchLastTieredOffsetEnable)
+
+ // Update per broker config
+ props.put(ReplicationConfigs.FOLLOWER_FETCH_LAST_TIERED_OFFSET_ENABLE_CONFIG, "false")
+ config.dynamicConfig.updateBrokerConfig(0, props)
+ assertFalse(config.followerFetchLastTieredOffsetEnable)
+ }
+
def verifyIncorrectLogLocalRetentionProps(logLocalRetentionMs: Long,
retentionMs: Long,
logLocalRetentionBytes: Long,
diff --git a/server/src/main/java/org/apache/kafka/server/config/ReplicationConfigs.java b/server/src/main/java/org/apache/kafka/server/config/ReplicationConfigs.java
index e437f6d4a76..d39a141a458 100644
--- a/server/src/main/java/org/apache/kafka/server/config/ReplicationConfigs.java
+++ b/server/src/main/java/org/apache/kafka/server/config/ReplicationConfigs.java
@@ -132,6 +132,10 @@ public class ReplicationConfigs {
public static final String REPLICA_SELECTOR_CLASS_CONFIG = "replica.selector.class";
public static final String REPLICA_SELECTOR_CLASS_DOC = "The fully qualified class name that implements ReplicaSelector. This is used by the broker to find the preferred read replica. By default, we use an implementation that returns the leader.";
+ public static final String FOLLOWER_FETCH_LAST_TIERED_OFFSET_ENABLE_CONFIG = "follower.fetch.last.tiered.offset.enable";
+ public static final String FOLLOWER_FETCH_LAST_TIERED_OFFSET_ENABLE_DOC = "When enabled, an empty follower skips offsets up to the last tiered offset and begins replication from the next offset.";
+ public static final boolean FOLLOWER_FETCH_LAST_TIERED_OFFSET_ENABLE_DEFAULT = false;
+
public static final String AUTO_LEADER_REBALANCE_ENABLE_CONFIG = "auto.leader.rebalance.enable";
public static final boolean AUTO_LEADER_REBALANCE_ENABLE_DEFAULT = true;
public static final String AUTO_LEADER_REBALANCE_ENABLE_DOC = String.format("Enables auto leader balancing. A background thread checks the distribution of partition leaders at regular intervals, configurable by %s. If the leader is imbalanced, leader rebalance to the preferred leader for partitions is triggered.",
@@ -158,6 +162,7 @@ public class ReplicationConfigs {
.define(UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG, BOOLEAN, LogConfig.DEFAULT_UNCLEAN_LEADER_ELECTION_ENABLE, HIGH, UNCLEAN_LEADER_ELECTION_ENABLE_DOC)
.define(INTER_BROKER_SECURITY_PROTOCOL_CONFIG, STRING, INTER_BROKER_SECURITY_PROTOCOL_DEFAULT, ConfigDef.ValidString.in(Utils.enumOptions(SecurityProtocol.class)), MEDIUM, INTER_BROKER_SECURITY_PROTOCOL_DOC)
.define(INTER_BROKER_LISTENER_NAME_CONFIG, STRING, null, MEDIUM, INTER_BROKER_LISTENER_NAME_DOC)
- .define(REPLICA_SELECTOR_CLASS_CONFIG, STRING, null, MEDIUM, REPLICA_SELECTOR_CLASS_DOC);
+ .define(REPLICA_SELECTOR_CLASS_CONFIG, STRING, null, MEDIUM, REPLICA_SELECTOR_CLASS_DOC)
+ .define(FOLLOWER_FETCH_LAST_TIERED_OFFSET_ENABLE_CONFIG, BOOLEAN, FOLLOWER_FETCH_LAST_TIERED_OFFSET_ENABLE_DEFAULT, MEDIUM, FOLLOWER_FETCH_LAST_TIERED_OFFSET_ENABLE_DOC);
}