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/01/19 11:05:23 UTC
(kafka) branch 4.0 updated: KAFKA-18589 Remove unused interBrokerProtocolVersion from GroupMetadataManager (#18619)
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 b31c357e73c KAFKA-18589 Remove unused interBrokerProtocolVersion from GroupMetadataManager (#18619)
b31c357e73c is described below
commit b31c357e73cd4b824659c63222487a87897b1ba1
Author: TengYao Chi <ki...@gmail.com>
AuthorDate: Sun Jan 19 18:59:24 2025 +0800
KAFKA-18589 Remove unused interBrokerProtocolVersion from GroupMetadataManager (#18619)
Reviewers: Chia-Ping Tsai <ch...@gmail.com>
---
core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala | 3 +--
.../main/scala/kafka/coordinator/group/GroupMetadataManager.scala | 3 +--
.../unit/kafka/coordinator/group/GroupMetadataManagerTest.scala | 7 +++----
3 files changed, 5 insertions(+), 8 deletions(-)
diff --git a/core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala b/core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala
index 2d317e4b55f..c1f071413e8 100644
--- a/core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala
+++ b/core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala
@@ -1804,8 +1804,7 @@ object GroupCoordinator {
groupMaxSize = config.groupCoordinatorConfig.classicGroupMaxSize,
groupInitialRebalanceDelayMs = config.groupCoordinatorConfig.classicGroupInitialRebalanceDelayMs)
- val groupMetadataManager = new GroupMetadataManager(config.brokerId, config.interBrokerProtocolVersion,
- offsetConfig, replicaManager, time, metrics)
+ val groupMetadataManager = new GroupMetadataManager(config.brokerId, offsetConfig, replicaManager, time, metrics)
new GroupCoordinator(config.brokerId, groupConfig, offsetConfig, groupMetadataManager, heartbeatPurgatory,
rebalancePurgatory, time, metrics)
}
diff --git a/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala b/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala
index d10e6852857..97ce088d48f 100644
--- a/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala
+++ b/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala
@@ -42,7 +42,7 @@ import org.apache.kafka.common.utils.{Time, Utils}
import org.apache.kafka.common.{TopicIdPartition, TopicPartition}
import org.apache.kafka.coordinator.group.{OffsetAndMetadata, OffsetConfig}
import org.apache.kafka.coordinator.group.generated.{GroupMetadataValue, OffsetCommitKey, OffsetCommitValue, GroupMetadataKey => GroupMetadataKeyData}
-import org.apache.kafka.server.common.{MetadataVersion, RequestLocal}
+import org.apache.kafka.server.common.RequestLocal
import org.apache.kafka.server.metrics.KafkaMetricsGroup
import org.apache.kafka.server.storage.log.FetchIsolation
import org.apache.kafka.server.util.KafkaScheduler
@@ -53,7 +53,6 @@ import scala.collection.mutable.ArrayBuffer
import scala.jdk.CollectionConverters._
class GroupMetadataManager(brokerId: Int,
- interBrokerProtocolVersion: MetadataVersion,
config: OffsetConfig,
val replicaManager: ReplicaManager,
time: Time,
diff --git a/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala b/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala
index b85160ce343..5897889ec3e 100644
--- a/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala
@@ -44,7 +44,7 @@ import org.apache.kafka.common.requests.ProduceResponse.PartitionResponse
import org.apache.kafka.common.utils.Utils
import org.apache.kafka.coordinator.group.{GroupCoordinatorConfig, OffsetAndMetadata, OffsetConfig}
import org.apache.kafka.coordinator.group.generated.{GroupMetadataValue, OffsetCommitValue}
-import org.apache.kafka.server.common.{MetadataVersion, RequestLocal}
+import org.apache.kafka.server.common.RequestLocal
import org.apache.kafka.server.metrics.KafkaYammerMetrics
import org.apache.kafka.server.storage.log.FetchIsolation
import org.apache.kafka.server.util.{KafkaScheduler, MockTime}
@@ -99,8 +99,7 @@ class GroupMetadataManagerTest {
metrics = new kMetrics()
time = new MockTime
replicaManager = mock(classOf[ReplicaManager])
- groupMetadataManager = new GroupMetadataManager(0, MetadataVersion.latestTesting, offsetConfig, replicaManager,
- time, metrics)
+ groupMetadataManager = new GroupMetadataManager(0, offsetConfig, replicaManager, time, metrics)
groupMetadataManager.startup(() => numOffsetsPartitions, enableMetadataExpiration = false)
partition = mock(classOf[Partition])
}
@@ -114,7 +113,7 @@ class GroupMetadataManagerTest {
def testLogInfoFromCleanupGroupMetadata(): Unit = {
var expiredOffsets: Int = 0
var infoCount = 0
- val gmm = new GroupMetadataManager(0, MetadataVersion.latestTesting, offsetConfig, replicaManager, time, metrics) {
+ val gmm = new GroupMetadataManager(0, offsetConfig, replicaManager, time, metrics) {
override def cleanupGroupMetadata(groups: Iterable[GroupMetadata], requestLocal: RequestLocal,
selector: GroupMetadata => Map[TopicPartition, OffsetAndMetadata]): Int = expiredOffsets