You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ch...@apache.org on 2025/02/10 23:31:21 UTC
(kafka) branch 4.0 updated: KAFKA-18366 Remove KafkaConfig.interBrokerProtocolVersion (#18820)
This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch 4.0
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/4.0 by this push:
new 8e8423fed03 KAFKA-18366 Remove KafkaConfig.interBrokerProtocolVersion (#18820)
8e8423fed03 is described below
commit 8e8423fed0358fdfe6baaa483c2c2bccb410b23c
Author: Ken Huang <s7...@gmail.com>
AuthorDate: Tue Feb 11 06:18:02 2025 +0800
KAFKA-18366 Remove KafkaConfig.interBrokerProtocolVersion (#18820)
Reviewers: Ismael Juma <is...@juma.me.uk>, Chia-Ping Tsai <ch...@gmail.com>
---
.../kafka/server/builders/LogManagerBuilder.java | 8 --------
core/src/main/scala/kafka/log/LogManager.scala | 9 +++------
core/src/main/scala/kafka/server/KafkaConfig.scala | 3 ---
.../unit/kafka/cluster/AbstractPartitionTest.scala | 2 +-
.../test/scala/unit/kafka/log/LogLoaderTest.scala | 1 -
.../test/scala/unit/kafka/log/LogManagerTest.scala | 2 --
.../scala/unit/kafka/server/KafkaConfigTest.scala | 9 ---------
.../unit/kafka/server/ReplicaFetcherThreadTest.scala | 18 +++++++++---------
.../scala/unit/kafka/server/ReplicaManagerTest.scala | 20 ++++++++++----------
core/src/test/scala/unit/kafka/utils/TestUtils.scala | 4 +---
.../jmh/fetcher/ReplicaFetcherThreadBenchmark.java | 5 ++---
.../partition/PartitionMakeFollowerBenchmark.java | 2 --
.../partition/UpdateFollowerFetchStateBenchmark.java | 2 --
.../org/apache/kafka/jmh/server/CheckpointBench.java | 3 +--
.../kafka/jmh/server/PartitionCreationBench.java | 2 --
15 files changed, 27 insertions(+), 63 deletions(-)
diff --git a/core/src/main/java/kafka/server/builders/LogManagerBuilder.java b/core/src/main/java/kafka/server/builders/LogManagerBuilder.java
index 44711760421..c70ee8f31b0 100644
--- a/core/src/main/java/kafka/server/builders/LogManagerBuilder.java
+++ b/core/src/main/java/kafka/server/builders/LogManagerBuilder.java
@@ -21,7 +21,6 @@ import kafka.log.LogManager;
import kafka.server.metadata.ConfigRepository;
import org.apache.kafka.common.utils.Time;
-import org.apache.kafka.server.common.MetadataVersion;
import org.apache.kafka.server.config.ServerLogConfigs;
import org.apache.kafka.server.util.Scheduler;
import org.apache.kafka.storage.internals.log.CleanerConfig;
@@ -51,7 +50,6 @@ public class LogManagerBuilder {
private long retentionCheckMs = 1000L;
private int maxTransactionTimeoutMs = 15 * 60 * 1000;
private ProducerStateManagerConfig producerStateManagerConfig = new ProducerStateManagerConfig(60000, false);
- private MetadataVersion interBrokerProtocolVersion = MetadataVersion.latestProduction();
private Scheduler scheduler = null;
private BrokerTopicStats brokerTopicStats = null;
private LogDirFailureChannel logDirFailureChannel = null;
@@ -120,11 +118,6 @@ public class LogManagerBuilder {
return this;
}
- public LogManagerBuilder setInterBrokerProtocolVersion(MetadataVersion interBrokerProtocolVersion) {
- this.interBrokerProtocolVersion = interBrokerProtocolVersion;
- return this;
- }
-
public LogManagerBuilder setScheduler(Scheduler scheduler) {
this.scheduler = scheduler;
return this;
@@ -181,7 +174,6 @@ public class LogManagerBuilder {
maxTransactionTimeoutMs,
producerStateManagerConfig,
PRODUCER_ID_EXPIRATION_CHECK_INTERVAL_MS,
- interBrokerProtocolVersion,
scheduler,
brokerTopicStats,
logDirFailureChannel,
diff --git a/core/src/main/scala/kafka/log/LogManager.scala b/core/src/main/scala/kafka/log/LogManager.scala
index 4c9c4eb475e..0ffe0e0488b 100755
--- a/core/src/main/scala/kafka/log/LogManager.scala
+++ b/core/src/main/scala/kafka/log/LogManager.scala
@@ -39,7 +39,6 @@ import org.apache.kafka.image.TopicsImage
import org.apache.kafka.metadata.properties.{MetaProperties, MetaPropertiesEnsemble, PropertiesUtils}
import java.util.{Collections, OptionalLong, Properties}
-import org.apache.kafka.server.common.MetadataVersion
import org.apache.kafka.server.metrics.KafkaMetricsGroup
import org.apache.kafka.server.util.{FileLock, Scheduler}
import org.apache.kafka.storage.internals.log.{CleanerConfig, LogConfig, LogDirFailureChannel, ProducerStateManagerConfig, RemoteIndexCache}
@@ -72,7 +71,6 @@ class LogManager(logDirs: Seq[File],
val maxTransactionTimeoutMs: Int,
val producerStateManagerConfig: ProducerStateManagerConfig,
val producerIdExpirationCheckIntervalMs: Int,
- interBrokerProtocolVersion: MetadataVersion,
scheduler: Scheduler,
brokerTopicStats: BrokerTopicStats,
logDirFailureChannel: LogDirFailureChannel,
@@ -507,7 +505,7 @@ class LogManager(logDirs: Seq[File],
}
if (remainingLogs == 0) {
- // loadLog is completed for all logs under the logDdir, mark it.
+ // loadLog is completed for all logs under the logDir, mark it.
loadLogsCompletedFlags.put(logDirAbsolutePath, true)
}
}
@@ -1002,7 +1000,7 @@ class LogManager(logDirs: Seq[File],
* @param isNew Whether the replica should have existed on the broker or not
* @param isFuture True if the future log of the specified partition should be returned or created
* @param topicId The topic ID of the partition's topic
- * @param targetLogDirectoryId The directory Id that should host the the partition's topic.
+ * @param targetLogDirectoryId The directory Id that should host the partition's topic.
* The next selected directory will be picked up if it None or equal {@link DirectoryId.UNASSIGNED}.
* The method assumes provided Id belong to online directory.
* @throws KafkaStorageException if isNew=false, log is not found in the cache and there is offline log directory on the broker
@@ -1318,7 +1316,7 @@ class LogManager(logDirs: Seq[File],
* deletion. Checkpoints are updated once all the directories have been renamed.
*
* @param topicPartitions The set of topic-partitions to delete asynchronously
- * @param errorHandler The error handler that will be called when a exception for a particular
+ * @param errorHandler The error handler that will be called when an exception for a particular
* topic-partition is raised
*/
def asyncDelete(topicPartitions: Iterable[TopicPartition],
@@ -1568,7 +1566,6 @@ object LogManager {
logDirFailureChannel = logDirFailureChannel,
time = time,
keepPartitionMetadataFile = keepPartitionMetadataFile,
- interBrokerProtocolVersion = config.interBrokerProtocolVersion,
remoteStorageSystemEnable = config.remoteLogManagerConfig.isRemoteStorageSystemEnabled(),
initialTaskDelayMs = config.logInitialTaskDelayMs)
}
diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala
index 046fe29d5e7..6698d9eb21f 100755
--- a/core/src/main/scala/kafka/server/KafkaConfig.scala
+++ b/core/src/main/scala/kafka/server/KafkaConfig.scala
@@ -388,9 +388,6 @@ class KafkaConfig private(doLog: Boolean, val props: util.Map[_, _])
val uncleanLeaderElectionCheckIntervalMs: Long = getLong(ReplicationConfigs.UNCLEAN_LEADER_ELECTION_INTERVAL_MS_CONFIG)
def uncleanLeaderElectionEnable: java.lang.Boolean = getBoolean(ReplicationConfigs.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG)
- // This will be removed soon. See KAFKA-18366.
- val interBrokerProtocolVersion = MetadataVersion.MINIMUM_KRAFT_VERSION
-
/** ********* Controlled shutdown configuration ***********/
val controlledShutdownEnable = getBoolean(ServerConfigs.CONTROLLED_SHUTDOWN_ENABLE_CONFIG)
diff --git a/core/src/test/scala/unit/kafka/cluster/AbstractPartitionTest.scala b/core/src/test/scala/unit/kafka/cluster/AbstractPartitionTest.scala
index 2cd271e1fe4..9d60328d868 100644
--- a/core/src/test/scala/unit/kafka/cluster/AbstractPartitionTest.scala
+++ b/core/src/test/scala/unit/kafka/cluster/AbstractPartitionTest.scala
@@ -77,7 +77,7 @@ class AbstractPartitionTest {
logDir1 = TestUtils.randomPartitionLogDir(tmpDir)
logDir2 = TestUtils.randomPartitionLogDir(tmpDir)
logManager = TestUtils.createLogManager(Seq(logDir1, logDir2), logConfig, configRepository,
- new CleanerConfig(false), time, interBrokerProtocolVersion, transactionVerificationEnabled = true)
+ new CleanerConfig(false), time, transactionVerificationEnabled = true)
logManager.startup(Set.empty)
alterPartitionManager = TestUtils.createAlterIsrManager()
diff --git a/core/src/test/scala/unit/kafka/log/LogLoaderTest.scala b/core/src/test/scala/unit/kafka/log/LogLoaderTest.scala
index 6d82ae2d291..b327c3b39e7 100644
--- a/core/src/test/scala/unit/kafka/log/LogLoaderTest.scala
+++ b/core/src/test/scala/unit/kafka/log/LogLoaderTest.scala
@@ -122,7 +122,6 @@ class LogLoaderTest {
maxTransactionTimeoutMs = maxTransactionTimeoutMs,
producerStateManagerConfig = producerStateManagerConfig,
producerIdExpirationCheckIntervalMs = producerIdExpirationCheckIntervalMs,
- interBrokerProtocolVersion = config.interBrokerProtocolVersion,
scheduler = time.scheduler,
brokerTopicStats = new BrokerTopicStats(),
logDirFailureChannel = logDirFailureChannel,
diff --git a/core/src/test/scala/unit/kafka/log/LogManagerTest.scala b/core/src/test/scala/unit/kafka/log/LogManagerTest.scala
index dcf3aaf3d0f..c0e4eaf5b46 100755
--- a/core/src/test/scala/unit/kafka/log/LogManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/log/LogManagerTest.scala
@@ -29,7 +29,6 @@ import org.apache.kafka.coordinator.transaction.TransactionLogConfig
import org.apache.kafka.image.{TopicImage, TopicsImage}
import org.apache.kafka.metadata.{LeaderRecoveryState, PartitionRegistration}
import org.apache.kafka.metadata.properties.{MetaProperties, MetaPropertiesEnsemble, MetaPropertiesVersion, PropertiesUtils}
-import org.apache.kafka.server.common.MetadataVersion
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.{AfterEach, BeforeEach, Test}
import org.mockito.ArgumentMatchers.any
@@ -1340,7 +1339,6 @@ class LogManagerTest {
brokerTopicStats = new BrokerTopicStats,
logDirFailureChannel = new LogDirFailureChannel(1),
keepPartitionMetadataFile = true,
- interBrokerProtocolVersion = MetadataVersion.latestTesting,
remoteStorageSystemEnable = false,
initialTaskDelayMs = 0)
diff --git a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
index 771ae409af8..5c95f409846 100755
--- a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
+++ b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
@@ -37,7 +37,6 @@ import org.apache.kafka.coordinator.group.modern.share.ShareGroupConfig
import org.apache.kafka.coordinator.transaction.{TransactionLogConfig, TransactionStateManagerConfig}
import org.apache.kafka.network.SocketServerConfigs
import org.apache.kafka.raft.QuorumConfig
-import org.apache.kafka.server.common.MetadataVersion
import org.apache.kafka.server.config.{DelegationTokenManagerConfigs, KRaftConfigs, QuotaConfig, ReplicationConfigs, ServerConfigs, ServerLogConfigs, ServerTopicConfigSynonyms}
import org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig
import org.apache.kafka.server.metrics.MetricConfigs
@@ -1591,14 +1590,6 @@ class KafkaConfigTest {
assertThrows(classOf[ConfigException], () => new KafkaConfig(props)).getMessage)
}
- @Test
- def testDefaultInterBrokerProtocolVersionKRaft(): Unit = {
- val props = new Properties()
- props.putAll(kraftProps())
- val config = KafkaConfig.fromProps(props)
- assertEquals(config.interBrokerProtocolVersion, MetadataVersion.MINIMUM_KRAFT_VERSION)
- }
-
@Test
def testMetadataMaxSnapshotInterval(): Unit = {
val validValue = 100
diff --git a/core/src/test/scala/unit/kafka/server/ReplicaFetcherThreadTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaFetcherThreadTest.scala
index 6bd3726ebd0..302ff2f39dd 100644
--- a/core/src/test/scala/unit/kafka/server/ReplicaFetcherThreadTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ReplicaFetcherThreadTest.scala
@@ -90,7 +90,7 @@ class ReplicaFetcherThreadTest {
val logContext = new LogContext(s"[ReplicaFetcher replicaId=${brokerConfig.brokerId}, leaderId=${leaderEndpointBlockingSend.brokerEndPoint().id}, fetcherId=$fetcherId] ")
val fetchSessionHandler = new FetchSessionHandler(logContext, leaderEndpointBlockingSend.brokerEndPoint().id)
val leader = new RemoteLeaderEndPoint(logContext.logPrefix, leaderEndpointBlockingSend, fetchSessionHandler,
- brokerConfig, replicaMgr, quota, () => brokerConfig.interBrokerProtocolVersion, () => 1)
+ brokerConfig, replicaMgr, quota, () => MetadataVersion.MINIMUM_KRAFT_VERSION, () => 1)
new ReplicaFetcherThread(name,
leader,
brokerConfig,
@@ -280,9 +280,9 @@ class ReplicaFetcherThreadTest {
val logContext = new LogContext(s"[ReplicaFetcher replicaId=${config.brokerId}, leaderId=${brokerEndPoint.id}, fetcherId=0] ")
val fetchSessionHandler = new FetchSessionHandler(logContext, brokerEndPoint.id)
val leader = new RemoteLeaderEndPoint(logContext.logPrefix, mockNetwork, fetchSessionHandler, config,
- replicaManager, quota, () => config.interBrokerProtocolVersion, () => 1)
+ replicaManager, quota, () => MetadataVersion.MINIMUM_KRAFT_VERSION, () => 1)
val thread = new ReplicaFetcherThread("bob", leader, config, failedPartitions,
- replicaManager, quota, logContext.logPrefix, () => config.interBrokerProtocolVersion) {
+ replicaManager, quota, logContext.logPrefix, () => MetadataVersion.MINIMUM_KRAFT_VERSION) {
override def processPartitionData(topicPartition: TopicPartition, fetchOffset: Long, partitionData: FetchData): Option[LogAppendInfo] = None
}
thread.addPartitions(Map(t1p0 -> initialFetchState(Some(topicId1), initialLEO), t1p1 -> initialFetchState(Some(topicId1), initialLEO)))
@@ -396,7 +396,7 @@ class ReplicaFetcherThreadTest {
config,
replicaManager,
quota,
- () => config.interBrokerProtocolVersion,
+ () => MetadataVersion.MINIMUM_KRAFT_VERSION,
() => 1
)
@@ -408,7 +408,7 @@ class ReplicaFetcherThreadTest {
replicaManager,
quota,
logContext.logPrefix,
- () => config.interBrokerProtocolVersion
+ () => MetadataVersion.MINIMUM_KRAFT_VERSION
)
thread.addPartitions(Map(
@@ -489,7 +489,7 @@ class ReplicaFetcherThreadTest {
config,
replicaManager,
quota,
- () => config.interBrokerProtocolVersion,
+ () => MetadataVersion.MINIMUM_KRAFT_VERSION,
() => 1
)
@@ -501,7 +501,7 @@ class ReplicaFetcherThreadTest {
replicaManager,
quota,
logContext.logPrefix,
- () => config.interBrokerProtocolVersion
+ () => MetadataVersion.MINIMUM_KRAFT_VERSION
)
thread.addPartitions(Map(
@@ -598,7 +598,7 @@ class ReplicaFetcherThreadTest {
val logContext = new LogContext(s"[ReplicaFetcher replicaId=${config.brokerId}, leaderId=${brokerEndPoint.id}, fetcherId=0] ")
val fetchSessionHandler = new FetchSessionHandler(logContext, brokerEndPoint.id)
val leader = new RemoteLeaderEndPoint(logContext.logPrefix, mockBlockingSend, fetchSessionHandler, config,
- replicaManager, replicaQuota, () => config.interBrokerProtocolVersion, () => 1)
+ replicaManager, replicaQuota, () => MetadataVersion.MINIMUM_KRAFT_VERSION, () => 1)
val thread = new ReplicaFetcherThread("bob",
leader,
config,
@@ -606,7 +606,7 @@ class ReplicaFetcherThreadTest {
replicaManager,
replicaQuota,
logContext.logPrefix,
- () => config.interBrokerProtocolVersion)
+ () => MetadataVersion.MINIMUM_KRAFT_VERSION)
val leaderEpoch = 1
diff --git a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
index a3095cfeb2c..63df3bfc8ee 100644
--- a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
@@ -271,7 +271,7 @@ class ReplicaManagerTest {
val logManager = TestUtils.createLogManager(config.logDirs.map(new File(_)), new LogConfig(new Properties()))
val metadataCache: MetadataCache = mock(classOf[MetadataCache])
mockGetAliveBrokerFunctions(metadataCache, Seq(new Node(0, "host0", 0)))
- when(metadataCache.metadataVersion()).thenReturn(config.interBrokerProtocolVersion)
+ when(metadataCache.metadataVersion()).thenReturn(MetadataVersion.MINIMUM_KRAFT_VERSION)
val rm = new ReplicaManager(
metrics = metrics,
config = config,
@@ -333,7 +333,7 @@ class ReplicaManagerTest {
val spyLogManager = spy(logManager)
val metadataCache: MetadataCache = mock(classOf[MetadataCache])
mockGetAliveBrokerFunctions(metadataCache, Seq(new Node(0, "host0", 0)))
- when(metadataCache.metadataVersion()).thenReturn(config.interBrokerProtocolVersion)
+ when(metadataCache.metadataVersion()).thenReturn(MetadataVersion.MINIMUM_KRAFT_VERSION)
val tp0 = new TopicPartition(topic, 0)
val uuid = Uuid.randomUuid()
val rm = new ReplicaManager(
@@ -406,7 +406,7 @@ class ReplicaManagerTest {
val aliveBrokers = Seq(new Node(0, "host0", 0), new Node(1, "host1", 1))
val metadataCache: MetadataCache = mock(classOf[MetadataCache])
mockGetAliveBrokerFunctions(metadataCache, aliveBrokers)
- when(metadataCache.metadataVersion()).thenReturn(config.interBrokerProtocolVersion)
+ when(metadataCache.metadataVersion()).thenReturn(MetadataVersion.MINIMUM_KRAFT_VERSION)
val rm = new ReplicaManager(
metrics = metrics,
config = config,
@@ -2765,7 +2765,7 @@ class ReplicaManagerTest {
any[TopicPartition], any[ListenerName])).
thenReturn(Map(leaderBrokerId -> new Node(leaderBrokerId, "host1", 9092, "rack-a"),
followerBrokerId -> new Node(followerBrokerId, "host2", 9092, "rack-b")).toMap)
- when(metadataCache.metadataVersion()).thenReturn(config.interBrokerProtocolVersion)
+ when(metadataCache.metadataVersion()).thenReturn(MetadataVersion.MINIMUM_KRAFT_VERSION)
val mockProducePurgatory = new DelayedOperationPurgatory[DelayedProduce](
"Produce", timer, 0, false)
val mockFetchPurgatory = new DelayedOperationPurgatory[DelayedFetch](
@@ -2818,9 +2818,9 @@ class ReplicaManagerTest {
s"fetcherId=$fetcherId] ")
val fetchSessionHandler = new FetchSessionHandler(logContext, sourceBroker.id)
val leader = new RemoteLeaderEndPoint(logContext.logPrefix, blockingSend, fetchSessionHandler, rm.config,
- rm, quotaManager.follower, () => rm.config.interBrokerProtocolVersion, () => 1)
+ rm, quotaManager.follower, () => MetadataVersion.MINIMUM_KRAFT_VERSION, () => 1)
new ReplicaFetcherThread(s"ReplicaFetcherThread-$fetcherId", leader, rm.config, failedPartitions, rm,
- quotaManager.follower, logContext.logPrefix, () => rm.config.interBrokerProtocolVersion) {
+ quotaManager.follower, logContext.logPrefix, () => MetadataVersion.MINIMUM_KRAFT_VERSION) {
override def doWork(): Unit = {
// In case the thread starts before the partition is added by AbstractFetcherManager,
// add it here (it's a no-op if already added)
@@ -3194,7 +3194,7 @@ class ReplicaManagerTest {
when(metadataCache.topicIdInfo()).thenReturn((topicIds.asJava, topicNames.asJava))
when(metadataCache.topicNamesToIds()).thenReturn(topicIds.asJava)
when(metadataCache.topicIdsToNames()).thenReturn(topicNames.asJava)
- when(metadataCache.metadataVersion()).thenReturn(config.interBrokerProtocolVersion)
+ when(metadataCache.metadataVersion()).thenReturn(MetadataVersion.MINIMUM_KRAFT_VERSION)
mockGetAliveBrokerFunctions(metadataCache, aliveBrokers)
val mockProducePurgatory = new DelayedOperationPurgatory[DelayedProduce](
"Produce", timer, 0, false)
@@ -3282,7 +3282,7 @@ class ReplicaManagerTest {
leader.setReplicaPartitionStateCallback(tp => PartitionState(leaderEpoch = 0))
val fetcher = new ReplicaFetcherThread(threadName, leader, config, failedPartitions, replicaManager,
- quotaManager, "", () => config.interBrokerProtocolVersion)
+ quotaManager, "", () => MetadataVersion.MINIMUM_KRAFT_VERSION)
val initialFetchState = InitialFetchState(
topicId = Some(Uuid.randomUuid()),
@@ -3529,8 +3529,8 @@ class ReplicaManagerTest {
val aliveBrokers = Seq(new Node(0, "host0", 0), new Node(1, "host1", 1))
mockGetAliveBrokerFunctions(metadataCache0, aliveBrokers)
mockGetAliveBrokerFunctions(metadataCache1, aliveBrokers)
- when(metadataCache0.metadataVersion()).thenReturn(config0.interBrokerProtocolVersion)
- when(metadataCache1.metadataVersion()).thenReturn(config1.interBrokerProtocolVersion)
+ when(metadataCache0.metadataVersion()).thenReturn(MetadataVersion.MINIMUM_KRAFT_VERSION)
+ when(metadataCache1.metadataVersion()).thenReturn(MetadataVersion.MINIMUM_KRAFT_VERSION)
// each replica manager is for a broker
val rm0 = new ReplicaManager(
diff --git a/core/src/test/scala/unit/kafka/utils/TestUtils.scala b/core/src/test/scala/unit/kafka/utils/TestUtils.scala
index 4e1b17c58a2..7be503fdbe0 100755
--- a/core/src/test/scala/unit/kafka/utils/TestUtils.scala
+++ b/core/src/test/scala/unit/kafka/utils/TestUtils.scala
@@ -52,7 +52,7 @@ import org.apache.kafka.network.SocketServerConfigs
import org.apache.kafka.network.metrics.RequestChannelMetrics
import org.apache.kafka.raft.QuorumConfig
import org.apache.kafka.server.authorizer.{AuthorizableRequestContext, Authorizer => JAuthorizer}
-import org.apache.kafka.server.common.{ControllerRequestCompletionHandler, MetadataVersion}
+import org.apache.kafka.server.common.ControllerRequestCompletionHandler
import org.apache.kafka.server.config.{DelegationTokenManagerConfigs, KRaftConfigs, ReplicationConfigs, ServerConfigs, ServerLogConfigs}
import org.apache.kafka.server.metrics.KafkaYammerMetrics
import org.apache.kafka.server.util.MockTime
@@ -960,7 +960,6 @@ object TestUtils extends Logging {
configRepository: ConfigRepository = new MockConfigRepository,
cleanerConfig: CleanerConfig = new CleanerConfig(false),
time: MockTime = new MockTime(),
- interBrokerProtocolVersion: MetadataVersion = MetadataVersion.latestTesting,
recoveryThreadsPerDataDir: Int = 4,
transactionVerificationEnabled: Boolean = false,
log: Option[UnifiedLog] = None,
@@ -984,7 +983,6 @@ object TestUtils extends Logging {
brokerTopicStats = new BrokerTopicStats,
logDirFailureChannel = new LogDirFailureChannel(logDirs.size),
keepPartitionMetadataFile = true,
- interBrokerProtocolVersion = interBrokerProtocolVersion,
remoteStorageSystemEnable = remoteStorageSystemEnable,
initialTaskDelayMs = initialTaskDelayMs)
diff --git a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/fetcher/ReplicaFetcherThreadBenchmark.java b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/fetcher/ReplicaFetcherThreadBenchmark.java
index 3b0afed5000..ab26b1deb31 100644
--- a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/fetcher/ReplicaFetcherThreadBenchmark.java
+++ b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/fetcher/ReplicaFetcherThreadBenchmark.java
@@ -141,7 +141,6 @@ public class ReplicaFetcherThreadBenchmark {
setFlushStartOffsetCheckpointMs(10000L).
setRetentionCheckMs(1000L).
setProducerStateManagerConfig(60000, false).
- setInterBrokerProtocolVersion(MetadataVersion.latestTesting()).
setScheduler(scheduler).
setBrokerTopicStats(brokerTopicStats).
setLogDirFailureChannel(logDirFailureChannel).
@@ -272,7 +271,7 @@ public class ReplicaFetcherThreadBenchmark {
config,
replicaManager,
replicaQuota,
- config::interBrokerProtocolVersion,
+ () -> MetadataVersion.MINIMUM_KRAFT_VERSION,
() -> -1L
) {
@Override
@@ -305,7 +304,7 @@ public class ReplicaFetcherThreadBenchmark {
replicaManager,
replicaQuota,
String.format("[ReplicaFetcher replicaId=%d, leaderId=%d, fetcherId=%d", config.brokerId(), 3, 3),
- config::interBrokerProtocolVersion
+ () -> MetadataVersion.MINIMUM_KRAFT_VERSION
);
pool = partitions;
diff --git a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/partition/PartitionMakeFollowerBenchmark.java b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/partition/PartitionMakeFollowerBenchmark.java
index 16ef047ca27..80505d53d29 100644
--- a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/partition/PartitionMakeFollowerBenchmark.java
+++ b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/partition/PartitionMakeFollowerBenchmark.java
@@ -34,7 +34,6 @@ import org.apache.kafka.common.record.SimpleRecord;
import org.apache.kafka.common.requests.LeaderAndIsrRequest;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.common.utils.Utils;
-import org.apache.kafka.server.common.MetadataVersion;
import org.apache.kafka.server.util.KafkaScheduler;
import org.apache.kafka.storage.internals.checkpoint.OffsetCheckpoints;
import org.apache.kafka.storage.internals.log.CleanerConfig;
@@ -111,7 +110,6 @@ public class PartitionMakeFollowerBenchmark {
setFlushStartOffsetCheckpointMs(10000L).
setRetentionCheckMs(1000L).
setProducerStateManagerConfig(60000, false).
- setInterBrokerProtocolVersion(MetadataVersion.latestTesting()).
setScheduler(scheduler).
setBrokerTopicStats(brokerTopicStats).
setLogDirFailureChannel(logDirFailureChannel).
diff --git a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/partition/UpdateFollowerFetchStateBenchmark.java b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/partition/UpdateFollowerFetchStateBenchmark.java
index 0fb7206301c..0706c6ee8ba 100644
--- a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/partition/UpdateFollowerFetchStateBenchmark.java
+++ b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/partition/UpdateFollowerFetchStateBenchmark.java
@@ -31,7 +31,6 @@ import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.requests.LeaderAndIsrRequest;
import org.apache.kafka.common.utils.Time;
-import org.apache.kafka.server.common.MetadataVersion;
import org.apache.kafka.server.util.KafkaScheduler;
import org.apache.kafka.storage.internals.checkpoint.OffsetCheckpoints;
import org.apache.kafka.storage.internals.log.CleanerConfig;
@@ -101,7 +100,6 @@ public class UpdateFollowerFetchStateBenchmark {
setFlushStartOffsetCheckpointMs(10000L).
setRetentionCheckMs(1000L).
setProducerStateManagerConfig(60000, false).
- setInterBrokerProtocolVersion(MetadataVersion.latestTesting()).
setScheduler(scheduler).
setBrokerTopicStats(brokerTopicStats).
setLogDirFailureChannel(logDirFailureChannel).
diff --git a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/server/CheckpointBench.java b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/server/CheckpointBench.java
index ddea968215e..25c11ca2057 100644
--- a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/server/CheckpointBench.java
+++ b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/server/CheckpointBench.java
@@ -31,7 +31,6 @@ import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.utils.Utils;
-import org.apache.kafka.server.common.MetadataVersion;
import org.apache.kafka.server.config.ServerLogConfigs;
import org.apache.kafka.server.util.KafkaScheduler;
import org.apache.kafka.server.util.MockTime;
@@ -112,7 +111,7 @@ public class CheckpointBench {
CollectionConverters.asJava(brokerProperties.logDirs()).stream().map(File::new).collect(Collectors.toList());
this.logManager = TestUtils.createLogManager(CollectionConverters.asScala(files),
new LogConfig(new Properties()), new MockConfigRepository(), new CleanerConfig(1, 4 * 1024 * 1024L, 0.9d,
- 1024 * 1024, 32 * 1024 * 1024, Double.MAX_VALUE, 15 * 1000, true), time, MetadataVersion.latestTesting(), 4, false, Option.empty(), false, ServerLogConfigs.LOG_INITIAL_TASK_DELAY_MS_DEFAULT);
+ 1024 * 1024, 32 * 1024 * 1024, Double.MAX_VALUE, 15 * 1000, true), time, 4, false, Option.empty(), false, ServerLogConfigs.LOG_INITIAL_TASK_DELAY_MS_DEFAULT);
scheduler.startup();
final BrokerTopicStats brokerTopicStats = new BrokerTopicStats(false);
final MetadataCache metadataCache =
diff --git a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/server/PartitionCreationBench.java b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/server/PartitionCreationBench.java
index 1ced9ba50b0..ceb7f8daebb 100644
--- a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/server/PartitionCreationBench.java
+++ b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/server/PartitionCreationBench.java
@@ -35,7 +35,6 @@ import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.requests.LeaderAndIsrRequest;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.common.utils.Utils;
-import org.apache.kafka.server.common.MetadataVersion;
import org.apache.kafka.server.util.KafkaScheduler;
import org.apache.kafka.server.util.Scheduler;
import org.apache.kafka.storage.internals.checkpoint.OffsetCheckpoints;
@@ -136,7 +135,6 @@ public class PartitionCreationBench {
setFlushStartOffsetCheckpointMs(10000L).
setRetentionCheckMs(1000L).
setProducerStateManagerConfig(60000, false).
- setInterBrokerProtocolVersion(MetadataVersion.latestTesting()).
setScheduler(scheduler).
setBrokerTopicStats(brokerTopicStats).
setLogDirFailureChannel(failureChannel).