Skip to content

KAFKA-15784: Ensure atomicity of in memory update and write when transactionally committing offsets #14774

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
jolshan merged 26 commits into from
Dec 14, 2023
Merged

KAFKA-15784: Ensure atomicity of in memory update and write when transactionally committing offsets #14774

Show file tree
Hide file tree
Changes from 17 commits
Commits
Show all changes
26 commits
Select commit Hold shift + click to select a range
e894d84
Redo verification path
jolshan Nov 16, 2023
2a52318
Fix build issues
jolshan Nov 17, 2023
865078b
Fix tests
jolshan Nov 17, 2023
a20f238
Merge branch 'trunk' of github.com:apache/kafka into kafka-15784
jolshan Nov 17, 2023
9bac9a9
Fix test failures
jolshan Nov 17, 2023
b4a920b
Rewrite GroupCoordinator and GroupMetadataManager to handle checks fo…
jolshan Nov 29, 2023
05c0d28
Merge branch 'trunk' of github.com:apache/kafka into kafka-15784
jolshan Nov 29, 2023
7bc6d06
Update comments and method names
jolshan Nov 30, 2023
a471f04
Fix style issues and passing verification guards
jolshan Dec 4, 2023
7c01682
Clean up GroupCoordinator, GroupMetadataManger, and ReplicaManager code
jolshan Dec 6, 2023
965ec39
remove package private scoping that is not needed
jolshan Dec 6, 2023
8ecbe4d
Remove produce path refactor
jolshan Dec 7, 2023
99e8577
spacing cleanups
jolshan Dec 7, 2023
a83b136
space
jolshan Dec 7, 2023
56a44ba
remove simple fix
jolshan Dec 7, 2023
e6a14ca
cleanups
jolshan Dec 7, 2023
de5dfad
Update comments simplify locking
jolshan Dec 8, 2023
66fff82
applying cherrypick
jolshan Dec 9, 2023
046425c
Fix tests
jolshan Dec 9, 2023
2c1f6a9
add error conversion
jolshan Dec 9, 2023
3f6c044
add test
jolshan Dec 9, 2023
3a3160d
Fix error code
jolshan Dec 11, 2023
4f757e2
Fix other error
jolshan Dec 11, 2023
ac80ae0
More error fixes and other cleanups
jolshan Dec 12, 2023
13dd654
fix silly error mix up
jolshan Dec 12, 2023
7cad567
rename method
jolshan Dec 12, 2023
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
44 changes: 36 additions & 8 deletions core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,9 +29,11 @@ import org.apache.kafka.common.message.LeaveGroupRequestData.MemberIdentity
import org.apache.kafka.common.metrics.Metrics
import org.apache.kafka.common.metrics.stats.Meter
import org.apache.kafka.common.protocol.{ApiKeys, Errors}
import org.apache.kafka.common.record.RecordBatch
import org.apache.kafka.common.requests._
import org.apache.kafka.common.utils.Time
import org.apache.kafka.server.record.BrokerCompressionType
import org.apache.kafka.storage.internals.log.VerificationGuard

import scala.collection.{Map, Seq, Set, immutable, mutable}
import scala.math.max
Expand Down Expand Up @@ -909,8 +911,31 @@ private[group] class GroupCoordinator(
val group = groupManager.getGroup(groupId).getOrElse {
groupManager.addGroup(new GroupMetadata(groupId, Empty, time))
}
doTxnCommitOffsets(group, transactionalId, memberId, groupInstanceId, generationId, producerId, producerEpoch,
offsetMetadata, requestLocal, responseCallback)

val offsetTopicPartition = new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, partitionFor(group.groupId))

def postVerificationCallback(
error: Errors,
newRequestLocal: RequestLocal,
verificationGuard: VerificationGuard
): Unit = {
if (error != Errors.NONE) {
responseCallback(offsetMetadata.map { case (k, _) => k -> error })
} else {
doTxnCommitOffsets(group, memberId, groupInstanceId, generationId, producerId, producerEpoch,
offsetTopicPartition, offsetMetadata, newRequestLocal, responseCallback, Some(verificationGuard))
}
}

groupManager.replicaManager.maybeStartTransactionVerificationForPartition(

@hachikuji hachikuji Dec 12, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

The access to replicaManager here probably suggests that we probably should be going through GroupMetadataManager. We could expose a wrapped maybeStartTransactionVerificationForPartition from GroupMetadataManager instead. That might also help us encapsulate the error conversion a little better.

@jolshan jolshan Dec 12, 2023

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.

i was just about to push my change before I saw this comment. I will address this comment tomorrow.

topicPartition = offsetTopicPartition,
transactionalId,
producerId,
producerEpoch,
RecordBatch.NO_SEQUENCE,
requestLocal,
postVerificationCallback
)
}
}

Expand Down Expand Up @@ -952,15 +977,16 @@ private[group] class GroupCoordinator(
}

private def doTxnCommitOffsets(group: GroupMetadata,
transactionalId: String,
memberId: String,
groupInstanceId: Option[String],
generationId: Int,
producerId: Long,
producerEpoch: Short,
offsetTopicPartition: TopicPartition,
offsetMetadata: immutable.Map[TopicIdPartition, OffsetAndMetadata],
requestLocal: RequestLocal,
responseCallback: immutable.Map[TopicIdPartition, Errors] => Unit): Unit = {
responseCallback: immutable.Map[TopicIdPartition, Errors] => Unit,
verificationGuard: Option[VerificationGuard]): Unit = {
group.inLock {
val validationErrorOpt = validateOffsetCommit(
group,
Expand All @@ -973,8 +999,8 @@ private[group] class GroupCoordinator(
if (validationErrorOpt.isDefined) {
responseCallback(offsetMetadata.map { case (k, _) => k -> validationErrorOpt.get })
} else {
groupManager.storeOffsets(group, memberId, offsetMetadata, responseCallback, transactionalId, producerId,
producerEpoch, requestLocal)
groupManager.storeOffsets(group, memberId, offsetTopicPartition, offsetMetadata, responseCallback, producerId,
producerEpoch, requestLocal, verificationGuard)
}
}
}
Expand Down Expand Up @@ -1033,19 +1059,21 @@ private[group] class GroupCoordinator(
isTransactional = false
)

val offsetTopicPartition = new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, partitionFor(group.groupId))

if (validationErrorOpt.isDefined) {
responseCallback(offsetMetadata.map { case (k, _) => k -> validationErrorOpt.get })
} else {
group.currentState match {
case Empty =>
groupManager.storeOffsets(group, memberId, offsetMetadata, responseCallback)
groupManager.storeOffsets(group, memberId, offsetTopicPartition, offsetMetadata, responseCallback, verificationGuard = None)

case Stable | PreparingRebalance =>
// During PreparingRebalance phase, we still allow a commit request since we rely
// on heartbeat response to eventually notify the rebalance in progress signal to the consumer
val member = group.get(memberId)
completeAndScheduleNextHeartbeatExpiration(group, member)
groupManager.storeOffsets(group, memberId, offsetMetadata, responseCallback, requestLocal = requestLocal)
groupManager.storeOffsets(group, memberId, offsetTopicPartition, offsetMetadata, responseCallback, requestLocal = requestLocal, verificationGuard = None)

case CompletingRebalance =>
// We should not receive a commit request if the group has not completed rebalance;
Expand Down
Loading