From a7e03d51d3d318d6f2e1c3bb6a6ce84b332298ce Mon Sep 17 00:00:00 2001 From: Guozhang Wang Date: Mon, 12 Oct 2020 15:18:29 -0700 Subject: [PATCH 1/3] more log4j --- .../main/scala/kafka/coordinator/group/GroupCoordinator.scala | 2 ++ .../kafka/coordinator/transaction/TransactionCoordinator.scala | 2 ++ 2 files changed, 4 insertions(+) diff --git a/core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala b/core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala index 60eb8da6c7865..5b6795716912d 100644 --- a/core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala +++ b/core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala @@ -908,6 +908,7 @@ class GroupCoordinator(val brokerId: Int, * @param offsetTopicPartitionId The partition we are now leading */ def onElection(offsetTopicPartitionId: Int): Unit = { + info(s"Becoming the group coordinator for partition $offsetTopicPartitionId") groupManager.scheduleLoadGroupAndOffsets(offsetTopicPartitionId, onGroupLoaded) } @@ -917,6 +918,7 @@ class GroupCoordinator(val brokerId: Int, * @param offsetTopicPartitionId The partition we are no longer leading */ def onResignation(offsetTopicPartitionId: Int): Unit = { + info(s"Resigning the group coordinator for partition $offsetTopicPartitionId") groupManager.removeGroupsForPartition(offsetTopicPartitionId, onGroupUnloaded) } diff --git a/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala b/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala index ebc0c2a1ad561..eb64c7bead7b3 100644 --- a/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala +++ b/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala @@ -311,6 +311,7 @@ class TransactionCoordinator(brokerId: Int, * @param coordinatorEpoch The partition coordinator (or leader) epoch from the received LeaderAndIsr request */ def onElection(txnTopicPartitionId: Int, coordinatorEpoch: Int): Unit = { + info(s"Becoming the txn coordinator for partition $txnTopicPartitionId at epoch $coordinatorEpoch") // The operations performed during immigration must be resilient to any previous errors we saw or partial state we // left off during the unloading phase. Ensure we remove all associated state for this partition before we continue // loading it. @@ -329,6 +330,7 @@ class TransactionCoordinator(brokerId: Int, * are resigning after receiving a StopReplica request from the controller */ def onResignation(txnTopicPartitionId: Int, coordinatorEpoch: Option[Int]): Unit = { + info(s"Resigning the txn coordinator for partition $txnTopicPartitionId at epoch $coordinatorEpoch") coordinatorEpoch match { case Some(epoch) => txnManager.removeTransactionsForTxnTopicPartition(txnTopicPartitionId, epoch) From 10808ca01c3c2de974803b1bdf848a27a057f3de Mon Sep 17 00:00:00 2001 From: Guozhang Wang Date: Wed, 14 Oct 2020 22:22:40 -0700 Subject: [PATCH 2/3] github comments --- .../main/scala/kafka/coordinator/group/GroupCoordinator.scala | 4 ++-- .../scala/kafka/coordinator/group/GroupMetadataManager.scala | 2 ++ 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala b/core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala index 5b6795716912d..aaa6079d82dfb 100644 --- a/core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala +++ b/core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala @@ -908,7 +908,7 @@ class GroupCoordinator(val brokerId: Int, * @param offsetTopicPartitionId The partition we are now leading */ def onElection(offsetTopicPartitionId: Int): Unit = { - info(s"Becoming the group coordinator for partition $offsetTopicPartitionId") + info(s"Electing as the group coordinator for partition $offsetTopicPartitionId") groupManager.scheduleLoadGroupAndOffsets(offsetTopicPartitionId, onGroupLoaded) } @@ -918,7 +918,7 @@ class GroupCoordinator(val brokerId: Int, * @param offsetTopicPartitionId The partition we are no longer leading */ def onResignation(offsetTopicPartitionId: Int): Unit = { - info(s"Resigning the group coordinator for partition $offsetTopicPartitionId") + info(s"Resigning as the group coordinator for partition $offsetTopicPartitionId") groupManager.removeGroupsForPartition(offsetTopicPartitionId, onGroupUnloaded) } diff --git a/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala b/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala index d733f346f2e04..584fe1300528f 100644 --- a/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala +++ b/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala @@ -544,6 +544,7 @@ class GroupMetadataManager(brokerId: Int, private[group] def loadGroupsAndOffsets(topicPartition: TopicPartition, onGroupLoaded: GroupMetadata => Unit, startTimeMs: java.lang.Long): Unit = { try { val schedulerTimeMs = time.milliseconds() - startTimeMs + debug(s"Started loading offsets and group metadata from $topicPartition") doLoadGroupsAndOffsets(topicPartition, onGroupLoaded) val endTimeMs = time.milliseconds() val totalLoadingTimeMs = endTimeMs - startTimeMs @@ -759,6 +760,7 @@ class GroupMetadataManager(brokerId: Int, var numOffsetsRemoved = 0 var numGroupsRemoved = 0 + debug(s"Started unloading offsets and group metadata for $topicPartition") inLock(partitionLock) { // we need to guard the group removal in cache in the loading partition lock // to prevent coordinator's check-and-get-group race condition From fc2b6ea03067095ac4f0a9939ac3bd54f973ded6 Mon Sep 17 00:00:00 2001 From: Guozhang Wang Date: Thu, 15 Oct 2020 09:27:30 -0700 Subject: [PATCH 3/3] github comments --- .../main/scala/kafka/coordinator/group/GroupCoordinator.scala | 4 ++-- .../coordinator/transaction/TransactionCoordinator.scala | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala b/core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala index aaa6079d82dfb..c407368994a52 100644 --- a/core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala +++ b/core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala @@ -908,7 +908,7 @@ class GroupCoordinator(val brokerId: Int, * @param offsetTopicPartitionId The partition we are now leading */ def onElection(offsetTopicPartitionId: Int): Unit = { - info(s"Electing as the group coordinator for partition $offsetTopicPartitionId") + info(s"Elected as the group coordinator for partition $offsetTopicPartitionId") groupManager.scheduleLoadGroupAndOffsets(offsetTopicPartitionId, onGroupLoaded) } @@ -918,7 +918,7 @@ class GroupCoordinator(val brokerId: Int, * @param offsetTopicPartitionId The partition we are no longer leading */ def onResignation(offsetTopicPartitionId: Int): Unit = { - info(s"Resigning as the group coordinator for partition $offsetTopicPartitionId") + info(s"Resigned as the group coordinator for partition $offsetTopicPartitionId") groupManager.removeGroupsForPartition(offsetTopicPartitionId, onGroupUnloaded) } diff --git a/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala b/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala index eb64c7bead7b3..147ff73629b8c 100644 --- a/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala +++ b/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala @@ -311,7 +311,7 @@ class TransactionCoordinator(brokerId: Int, * @param coordinatorEpoch The partition coordinator (or leader) epoch from the received LeaderAndIsr request */ def onElection(txnTopicPartitionId: Int, coordinatorEpoch: Int): Unit = { - info(s"Becoming the txn coordinator for partition $txnTopicPartitionId at epoch $coordinatorEpoch") + info(s"Elected as the txn coordinator for partition $txnTopicPartitionId at epoch $coordinatorEpoch") // The operations performed during immigration must be resilient to any previous errors we saw or partial state we // left off during the unloading phase. Ensure we remove all associated state for this partition before we continue // loading it. @@ -330,7 +330,7 @@ class TransactionCoordinator(brokerId: Int, * are resigning after receiving a StopReplica request from the controller */ def onResignation(txnTopicPartitionId: Int, coordinatorEpoch: Option[Int]): Unit = { - info(s"Resigning the txn coordinator for partition $txnTopicPartitionId at epoch $coordinatorEpoch") + info(s"Resigned as the txn coordinator for partition $txnTopicPartitionId at epoch $coordinatorEpoch") coordinatorEpoch match { case Some(epoch) => txnManager.removeTransactionsForTxnTopicPartition(txnTopicPartitionId, epoch)