Skip to content

KAFKA-10060 GroupMetadataManager should not log if there are no offse… #8767

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ijuma merged 1 commit into from
May 31, 2020
Merged

KAFKA-10060 GroupMetadataManager should not log if there are no offse… #8767

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -786,7 +786,8 @@ class GroupMetadataManager(brokerId: Int,
group.removeExpiredOffsets(currentTimestamp, config.offsetsRetentionMs)
})
offsetExpiredSensor.record(numOffsetsRemoved)
info(s"Removed $numOffsetsRemoved expired offsets in ${time.milliseconds() - currentTimestamp} milliseconds.")
if (numOffsetsRemoved > 0)
info(s"Removed $numOffsetsRemoved expired offsets in ${time.milliseconds() - currentTimestamp} milliseconds.")
}

/**
Expand Down
41 changes: 32 additions & 9 deletions core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,7 +57,6 @@ class GroupMetadataManagerTest {
var replicaManager: ReplicaManager = null
var groupMetadataManager: GroupMetadataManager = null
var scheduler: KafkaScheduler = null
var zkClient: KafkaZkClient = null
var partition: Partition = null
var defaultOffsetRetentionMs = Long.MaxValue
var metrics: kMetrics = null
Expand All @@ -71,11 +70,9 @@ class GroupMetadataManagerTest {
val sessionTimeout = 10000
val defaultRequireStable = false

@Before
def setUp(): Unit = {
private val offsetConfig = {
val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(nodeId = 0, zkConnect = ""))

val offsetConfig = OffsetConfig(maxMetadataSize = config.offsetMetadataMaxSize,
OffsetConfig(maxMetadataSize = config.offsetMetadataMaxSize,
loadBufferSize = config.offsetsLoadBufferSize,
offsetsRetentionMs = config.offsetsRetentionMinutes * 60 * 1000L,
offsetsRetentionCheckIntervalMs = config.offsetsRetentionCheckIntervalMs,
Expand All @@ -85,21 +82,47 @@ class GroupMetadataManagerTest {
offsetsTopicCompressionCodec = config.offsetsTopicCompressionCodec,
offsetCommitTimeoutMs = config.offsetCommitTimeoutMs,
offsetCommitRequiredAcks = config.offsetCommitRequiredAcks)
}

defaultOffsetRetentionMs = offsetConfig.offsetsRetentionMs

private def mockKafkaZkClient: KafkaZkClient = {
// make two partitions of the group topic to make sure some partitions are not owned by the coordinator
zkClient = EasyMock.createNiceMock(classOf[KafkaZkClient])
val zkClient: KafkaZkClient = EasyMock.createNiceMock(classOf[KafkaZkClient])
EasyMock.expect(zkClient.getTopicPartitionCount(Topic.GROUP_METADATA_TOPIC_NAME)).andReturn(Some(2))
EasyMock.replay(zkClient)
zkClient
}

@Before
def setUp(): Unit = {
defaultOffsetRetentionMs = offsetConfig.offsetsRetentionMs
metrics = new kMetrics()
time = new MockTime
replicaManager = EasyMock.createNiceMock(classOf[ReplicaManager])
groupMetadataManager = new GroupMetadataManager(0, ApiVersion.latestVersion, offsetConfig, replicaManager, zkClient, time, metrics)
groupMetadataManager = new GroupMetadataManager(0, ApiVersion.latestVersion, offsetConfig, replicaManager,
mockKafkaZkClient, time, metrics)
partition = EasyMock.niceMock(classOf[Partition])
}

@Test
def testLogInfoFromCleanupGroupMetadata(): Unit = {
var expiredOffsets: Int = 0
var infoCount = 0
val gmm = new GroupMetadataManager(0, ApiVersion.latestVersion, offsetConfig, replicaManager, mockKafkaZkClient, time, metrics) {

@chia7712 chia7712 May 31, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

create another GroupMetadataManager to override the methods we want to verify

override def cleanupGroupMetadata(groups: Iterable[GroupMetadata],
selector: GroupMetadata => Map[TopicPartition, OffsetAndMetadata]): Int = expiredOffsets

override def info(msg: => String): Unit = infoCount += 1
}

// if there are no offsets to expire, we skip to log
gmm.cleanupGroupMetadata()
assertEquals(0, infoCount)
// if there are offsets to expire, we should log info
expiredOffsets = 100
gmm.cleanupGroupMetadata()
assertEquals(1, infoCount)
}

@Test
def testLoadOffsetsWithoutGroup(): Unit = {
val groupMetadataTopicPartition = groupTopicPartition
Expand Down