You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by jo...@apache.org on 2023/12/07 23:55:04 UTC
(kafka) 01/01: cleanups
This is an automated email from the ASF dual-hosted git repository.
jolshan pushed a commit to branch kafka-15784
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit e6a14cae826918abe07cdd3018d72024e5603fa1
Author: Justine <jo...@confluent.io>
AuthorDate: Thu Dec 7 15:54:41 2023 -0800
cleanups
---
.../scala/kafka/coordinator/group/GroupMetadataManager.scala | 11 ++++-------
core/src/main/scala/kafka/server/ReplicaManager.scala | 5 ++++-
2 files changed, 8 insertions(+), 8 deletions(-)
diff --git a/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala b/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala
index 06e1dbe9477..4a0b45dbe72 100644
--- a/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala
+++ b/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala
@@ -344,7 +344,7 @@ class GroupMetadataManager(brokerId: Int,
preAppendErrors = preAppendErrors)
}
- def generateOffsetRecords(magicValue: Byte,
+ private def generateOffsetRecords(magicValue: Byte,
isTxnOffsetCommit: Boolean,
groupId: String,
offsetTopicPartition: TopicPartition,
@@ -372,13 +372,13 @@ class GroupMetadataManager(brokerId: Int,
Map(offsetTopicPartition -> builder.build())
}
- def createPutCacheCallback(isTxnOffsetCommit: Boolean,
+ private def createPutCacheCallback(isTxnOffsetCommit: Boolean,
group: GroupMetadata,
consumerId: String,
offsetMetadata: immutable.Map[TopicIdPartition, OffsetAndMetadata],
filteredOffsetMetadata: Map[TopicIdPartition, OffsetAndMetadata],
responseCallback: immutable.Map[TopicIdPartition, Errors] => Unit,
- producerId: Long = RecordBatch.NO_PRODUCER_ID,
+ producerId: Long,
records: Map[TopicPartition, MemoryRecords],
preAppendErrors: Map[TopicPartition, LogAppendResult] = Map.empty): Map[TopicPartition, PartitionResponse] => Unit = {
val offsetTopicPartition = new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, partitionFor(group.groupId))
@@ -502,10 +502,7 @@ class GroupMetadataManager(brokerId: Int,
val records = generateOffsetRecords(magicOpt.get, isTxnOffsetCommit, group.groupId, offsetTopicPartition, filteredOffsetMetadata, producerId, producerEpoch)
val putCacheCallback = createPutCacheCallback(isTxnOffsetCommit, group, consumerId, offsetMetadata, filteredOffsetMetadata, responseCallback, producerId, records)
- val verificationGuards = verificationGuard match {
- case Some(guard) => Map(offsetTopicPartition -> guard)
- case None => Map.empty[TopicPartition, VerificationGuard]
- }
+ val verificationGuards = verificationGuard.map(guard => offsetTopicPartition -> guard).toMap
group.inLock {
if (isTxnOffsetCommit) {
diff --git a/core/src/main/scala/kafka/server/ReplicaManager.scala b/core/src/main/scala/kafka/server/ReplicaManager.scala
index bb8b8eff4a2..f3871718529 100644
--- a/core/src/main/scala/kafka/server/ReplicaManager.scala
+++ b/core/src/main/scala/kafka/server/ReplicaManager.scala
@@ -1106,7 +1106,10 @@ class ReplicaManager(val config: KafkaConfig,
def generalizedCallback(preAppendErrors: Map[TopicPartition, Errors],
newRequestLocal: RequestLocal,
verificationGuards: Map[TopicPartition, VerificationGuard]): Unit = {
- callback(preAppendErrors.getOrElse(topicPartition, Errors.NONE), newRequestLocal, verificationGuards.getOrElse(topicPartition, VerificationGuard.SENTINEL))
+ callback(
+ preAppendErrors.getOrElse(topicPartition, Errors.NONE),
+ newRequestLocal,
+ verificationGuards.getOrElse(topicPartition, VerificationGuard.SENTINEL))
}
maybeStartTransactionVerificationForPartitions(