dajac commented on code in PR #13787:
URL: https://github.com/apache/kafka/pull/13787#discussion_r1219799829
##########
core/src/main/scala/kafka/log/UnifiedLog.scala:
##########
@@ -579,9 +579,30 @@ class UnifiedLog(@volatile var logStartOffset: Long,
result
}
+ /**
+ * Maybe create and return the verificationGuard object for the given
producer ID if the transaction is not yet ongoing.
Review Comment:
ditto.
##########
core/src/main/scala/kafka/log/UnifiedLog.scala:
##########
@@ -980,6 +1005,26 @@ class UnifiedLog(@volatile var logStartOffset: Long,
if (duplicateBatch.isPresent) {
return (updatedProducers, completedTxns.toList,
Some(duplicateBatch.get()))
}
+
+ // Verify that if the record is transactional & the append origin is
client, that we either have an ongoing transaction or verified transaction
state.
+ // This guarantees that transactional records are never written to
the log outside of the transaction coordinator's knowledge of an open
transaction on
+ // the partition. If we do not have an ongoing transaction or
correct guard, return an error and do not append.
+ // There are two phases -- the first append to the log and
subsequent appends.
+ //
+ // 1. First append: Verification starts with creating a
verificationGuard, sending a verification request to the transaction
coordinator, and
Review Comment:
nit `verification guard` for consistency. The are a few other cases in this
comment.
##########
core/src/test/scala/unit/kafka/log/ProducerStateManagerTest.scala:
##########
@@ -1087,6 +1087,46 @@ class ProducerStateManagerTest {
assertTrue(!manager.latestSnapshotOffset.isPresent)
}
+ @Test
+ def testEntryForVerification(): Unit = {
+ val originalEntry = stateManager.verificationStateEntry(producerId, true)
+ val originalEntryVerificationGuard = originalEntry.verificationGuard()
+
+ def verifyEntry(producerId: Long, newEntry: VerificationStateEntry): Unit
= {
+ val entry = stateManager.verificationStateEntry(producerId, false)
+ assertEquals(originalEntryVerificationGuard, entry.verificationGuard)
+ assertEquals(entry.verificationGuard, newEntry.verificationGuard)
+ }
+
+ // If we already have an entry, reuse it.
+ val updatedEntry = stateManager.verificationStateEntry(producerId, true)
+ verifyEntry(producerId, updatedEntry)
+
+ // Before we add transactional data, we can't remove the entry.
+ stateManager.clearVerificationStateEntry(producerId)
+ verifyEntry(producerId, updatedEntry)
+
+ // Add the transactional data and clear the entry
+ append(stateManager, producerId, 0, 0, offset = 0, isTransactional = true)
+ stateManager.clearVerificationStateEntry(producerId)
+ assertEquals(null, stateManager.verificationStateEntry(producerId, false))
+
Review Comment:
nit: Empty line could be removed.
##########
core/src/main/scala/kafka/cluster/Partition.scala:
##########
@@ -575,8 +575,12 @@ class Partition(val topicPartition: TopicPartition,
}
}
- def hasOngoingTransaction(producerId: Long): Boolean = {
- leaderLogIfLocal.exists(leaderLog =>
leaderLog.hasOngoingTransaction(producerId))
+ // Returns a verificationGuard object if we need to verify. This starts or
continues the verification process. Otherwise return null.
Review Comment:
nit: `verification guard object`? I find `verificationGuard` confusing
because we actually return an `Object`.
##########
core/src/main/scala/kafka/log/UnifiedLog.scala:
##########
@@ -980,6 +1005,26 @@ class UnifiedLog(@volatile var logStartOffset: Long,
if (duplicateBatch.isPresent) {
return (updatedProducers, completedTxns.toList,
Some(duplicateBatch.get()))
}
+
+ // Verify that if the record is transactional & the append origin is
client, that we either have an ongoing transaction or verified transaction
state.
+ // This guarantees that transactional records are never written to
the log outside of the transaction coordinator's knowledge of an open
transaction on
+ // the partition. If we do not have an ongoing transaction or
correct guard, return an error and do not append.
+ // There are two phases -- the first append to the log and
subsequent appends.
+ //
+ // 1. First append: Verification starts with creating a
verificationGuard, sending a verification request to the transaction
coordinator, and
+ // given a "verified" response, continuing the append path. (A
non-verified response throws an error.) We create the unique verification guard
for the transaction
+ // to ensure there is no race between the transaction coordinator
response and an abort marker getting written to the log. We need a unique guard
because we could
+ // have a sequence of events where we start a transaction
verification, have the transaction coordinator send a verified response, write
an abort marker,
+ // start a new transaction not aware of the partition, and receive
the stale verification (ABA problem). With a unique verificationGuard, this
sequence would not
+ // result in appending to the log and would return an error. The
guard is removed after the first append to the transaction and from then, we
can rely on phase 2.
+ //
+ // 2. Subsequent appends: Once we write to the transaction, the
in-memory state currentTxnFirstOffset is populated. This field remains until the
+ // transaction is completed or aborted. We can guarantee the
transaction coordinator knows about the transaction given step 1 and that the
transaction is still
+ // ongoing. If the transaction is expected to be ongoing, we will
not st a verification guard. If the transaction is aborted,
hasOngoingTransaction is false and
Review Comment:
nit: s/st/set
##########
core/src/test/scala/unit/kafka/log/ProducerStateManagerTest.scala:
##########
@@ -1087,6 +1087,46 @@ class ProducerStateManagerTest {
assertTrue(!manager.latestSnapshotOffset.isPresent)
}
+ @Test
+ def testEntryForVerification(): Unit = {
+ val originalEntry = stateManager.verificationStateEntry(producerId, true)
+ val originalEntryVerificationGuard = originalEntry.verificationGuard()
+
+ def verifyEntry(producerId: Long, newEntry: VerificationStateEntry): Unit
= {
+ val entry = stateManager.verificationStateEntry(producerId, false)
+ assertEquals(originalEntryVerificationGuard, entry.verificationGuard)
+ assertEquals(entry.verificationGuard, newEntry.verificationGuard)
+ }
+
+ // If we already have an entry, reuse it.
+ val updatedEntry = stateManager.verificationStateEntry(producerId, true)
+ verifyEntry(producerId, updatedEntry)
+
+ // Before we add transactional data, we can't remove the entry.
+ stateManager.clearVerificationStateEntry(producerId)
+ verifyEntry(producerId, updatedEntry)
+
+ // Add the transactional data and clear the entry
Review Comment:
nit: `.`
##########
core/src/main/scala/kafka/log/UnifiedLog.scala:
##########
@@ -579,9 +579,30 @@ class UnifiedLog(@volatile var logStartOffset: Long,
result
}
+ /**
+ * Maybe create and return the verificationGuard object for the given
producer ID if the transaction is not yet ongoing.
+ * Creation starts the verification process. Otherwise return null.
+ */
+ def maybeStartTransactionVerification(producerId: Long): Object = lock
synchronized {
+ if (hasOngoingTransaction(producerId))
+ null
+ else
+ verificationGuard(producerId, true)
+ }
+
+ /**
+ * Maybe create the VerificationStateEntry for the given producer ID -- if
an entry is present, return its verification guard, otherwise, return null.
+ */
+ def verificationGuard(producerId: Long, createIfAbsent: Boolean = false):
Object = lock synchronized {
+ val entry = producerStateManager.verificationStateEntry(producerId,
createIfAbsent)
+ if (entry != null) entry.verificationGuard else null
+ }
+
+ /**
+ * Return true if the given producer ID has a transaction ongoing.
+ */
def hasOngoingTransaction(producerId: Long): Boolean = lock synchronized {
- val entry = producerStateManager.activeProducers.get(producerId)
- entry != null && entry.currentTxnFirstOffset.isPresent
+ producerStateManager.activeProducers.getOrDefault(producerId,
ProducerStateEntry.empty(producerId)).currentTxnFirstOffset.isPresent
Review Comment:
nit: The previous version seems better to me as we don't have to allocate
`ProducerStateEntry.empty(producerId)` all the time.
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/VerificationStateEntry.java:
##########
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.storage.internals.log;
+
+/**
+ * This class represents the verification state of a specific producer id.
+ * It contains a verificationGuard object that is used to uniquely identify
the transaction we want to verify.
+ * After verifying, we retain this object until we append to the log. This
prevents any race conditions where the transaction
+ * may end via a control marker before we write to the log. This mechanism is
used to prevent hanging transactions.
+ * We remove the verificationGuard whenever we write data to the transaction
or write an end marker for the transaction.
+ * Any lingering entries that are never verified are removed via the producer
state entry cleanup mechanism.
+ */
+public class VerificationStateEntry {
+
+ private long timestamp;
+ private Object verificationGuard;
Review Comment:
nit: Those two could be `final`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
