diff --git a/core/src/main/scala/kafka/cluster/Partition.scala b/core/src/main/scala/kafka/cluster/Partition.scala index e1478c62d510c..970f5146b7ccc 100755 --- a/core/src/main/scala/kafka/cluster/Partition.scala +++ b/core/src/main/scala/kafka/cluster/Partition.scala @@ -577,9 +577,9 @@ class Partition(val topicPartition: TopicPartition, } // Returns a verification guard object if we need to verify. This starts or continues the verification process. Otherwise return null. - def maybeStartTransactionVerification(producerId: Long): Object = { + def maybeStartTransactionVerification(producerId: Long, sequence: Int, epoch: Short): Object = { leaderLogIfLocal match { - case Some(log) => log.maybeStartTransactionVerification(producerId) + case Some(log) => log.maybeStartTransactionVerification(producerId, sequence, epoch) case None => throw new NotLeaderOrFollowerException(); } } diff --git a/core/src/main/scala/kafka/log/UnifiedLog.scala b/core/src/main/scala/kafka/log/UnifiedLog.scala index 9eb1af6d56fd2..f619873c4ac75 100644 --- a/core/src/main/scala/kafka/log/UnifiedLog.scala +++ b/core/src/main/scala/kafka/log/UnifiedLog.scala @@ -581,18 +581,27 @@ class UnifiedLog(@volatile var logStartOffset: Long, * Maybe create and return the verification guard 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 { + def maybeStartTransactionVerification(producerId: Long, sequence: Int, epoch: Short): Object = lock synchronized { if (hasOngoingTransaction(producerId)) null else - getOrMaybeCreateVerificationGuard(producerId, true) + maybeCreateVerificationGuard(producerId, sequence, epoch) } /** - * Maybe create the VerificationStateEntry for the given producer ID -- if an entry is present, return its verification guard, otherwise, return null. + * Maybe create the VerificationStateEntry for the given producer ID -- always return the verification guard */ - def getOrMaybeCreateVerificationGuard(producerId: Long, createIfAbsent: Boolean = false): Object = lock synchronized { - val entry = producerStateManager.verificationStateEntry(producerId, createIfAbsent) + def maybeCreateVerificationGuard(producerId: Long, + sequence: Int, + epoch: Short): Object = lock synchronized { + producerStateManager.maybeCreateVerificationStateEntry(producerId, sequence, epoch).verificationGuard + } + + /** + * If an VerificationStateEntry is present for the given producer ID, return its verification guard, otherwise, return null. + */ + def verificationGuard(producerId: Long): Object = lock synchronized { + val entry = producerStateManager.verificationStateEntry(producerId) if (entry != null) entry.verificationGuard else null } @@ -1042,7 +1051,8 @@ class UnifiedLog(@volatile var logStartOffset: Long, } private def batchMissingRequiredVerification(batch: MutableRecordBatch, requestVerificationGuard: Object): Boolean = { - producerStateManager.producerStateManagerConfig().transactionVerificationEnabled() && (requestVerificationGuard != getOrMaybeCreateVerificationGuard(batch.producerId) || requestVerificationGuard == null) + producerStateManager.producerStateManagerConfig().transactionVerificationEnabled() && + (requestVerificationGuard != verificationGuard(batch.producerId) || requestVerificationGuard == null) } /** diff --git a/core/src/main/scala/kafka/server/ReplicaManager.scala b/core/src/main/scala/kafka/server/ReplicaManager.scala index aea9adb721225..69d291f84fa7b 100644 --- a/core/src/main/scala/kafka/server/ReplicaManager.scala +++ b/core/src/main/scala/kafka/server/ReplicaManager.scala @@ -862,7 +862,8 @@ class ReplicaManager(val config: KafkaConfig, if (transactionalBatches.nonEmpty) { // We return verification guard if the partition needs to be verified. If no state is present, no need to verify. - val verificationGuard = getPartitionOrException(topicPartition).maybeStartTransactionVerification(records.firstBatch.producerId) + val firstBatch = records.firstBatch + val verificationGuard = getPartitionOrException(topicPartition).maybeStartTransactionVerification(firstBatch.producerId, firstBatch.baseSequence, firstBatch.producerEpoch) if (verificationGuard != null) { verificationGuards.put(topicPartition, verificationGuard) unverifiedEntries.put(topicPartition, records) diff --git a/core/src/test/scala/unit/kafka/cluster/PartitionTest.scala b/core/src/test/scala/unit/kafka/cluster/PartitionTest.scala index ab3d06ef65369..b451dd8578416 100644 --- a/core/src/test/scala/unit/kafka/cluster/PartitionTest.scala +++ b/core/src/test/scala/unit/kafka/cluster/PartitionTest.scala @@ -998,7 +998,7 @@ class PartitionTest extends AbstractPartitionTest { new SimpleRecord("k3".getBytes, "v3".getBytes)), baseOffset = 0L, producerId = 2L) - val verificationGuard = partition.maybeStartTransactionVerification(2L) + val verificationGuard = partition.maybeStartTransactionVerification(2L, 0, 0) partition.appendRecordsToLeader(records, origin = AppendOrigin.CLIENT, requiredAcks = 0, RequestLocal.withThreadConfinedCaching, verificationGuard) def fetchOffset(isolationLevel: Option[IsolationLevel], timestamp: Long): TimestampAndOffset = { @@ -3390,7 +3390,7 @@ class PartitionTest extends AbstractPartitionTest { assertThrows(classOf[InvalidRecordException], () => partition.appendRecordsToLeader(transactionRecords(), origin = AppendOrigin.CLIENT, requiredAcks = 1, RequestLocal.withThreadConfinedCaching)) // Before appendRecordsToLeader is called, ReplicaManager will call maybeStartTransactionVerification. We should get a non-null verification object. - val verificationGuard = partition.maybeStartTransactionVerification(producerId) + val verificationGuard = partition.maybeStartTransactionVerification(producerId, 3, 0) assertNotNull(verificationGuard) // With the wrong verification guard, append should fail. @@ -3398,12 +3398,12 @@ class PartitionTest extends AbstractPartitionTest { origin = AppendOrigin.CLIENT, requiredAcks = 1, RequestLocal.withThreadConfinedCaching, Optional.of(new Object))) // We should return the same verification object when we still need to verify. Append should proceed. - val verificationGuard2 = partition.maybeStartTransactionVerification(producerId) + val verificationGuard2 = partition.maybeStartTransactionVerification(producerId, 3, 0) assertEquals(verificationGuard, verificationGuard2) partition.appendRecordsToLeader(transactionRecords(), origin = AppendOrigin.CLIENT, requiredAcks = 1, RequestLocal.withThreadConfinedCaching, verificationGuard) // We should no longer need a verification object. Future appends without verification guard will also succeed. - val verificationGuard3 = partition.maybeStartTransactionVerification(producerId) + val verificationGuard3 = partition.maybeStartTransactionVerification(producerId, 3, 0) assertNull(verificationGuard3) partition.appendRecordsToLeader(transactionRecords(), origin = AppendOrigin.CLIENT, requiredAcks = 1, RequestLocal.withThreadConfinedCaching) } diff --git a/core/src/test/scala/unit/kafka/log/ProducerStateManagerTest.scala b/core/src/test/scala/unit/kafka/log/ProducerStateManagerTest.scala index 7f8d48071a6d2..64a8a5f72c7e3 100644 --- a/core/src/test/scala/unit/kafka/log/ProducerStateManagerTest.scala +++ b/core/src/test/scala/unit/kafka/log/ProducerStateManagerTest.scala @@ -200,7 +200,8 @@ class ProducerStateManagerTest { val producerEpoch = 0.toShort val offset = 992342L val seq = 0 - val producerAppendInfo = new ProducerAppendInfo(partition, producerId, ProducerStateEntry.empty(producerId), AppendOrigin.CLIENT) + val producerAppendInfo = new ProducerAppendInfo(partition, producerId, ProducerStateEntry.empty(producerId), AppendOrigin.CLIENT, + stateManager.maybeCreateVerificationStateEntry(producerId, seq, producerEpoch)) val firstOffsetMetadata = new LogOffsetMetadata(offset, 990000L, 234224) producerAppendInfo.appendDataBatch(producerEpoch, seq, seq, time.milliseconds(), @@ -388,7 +389,8 @@ class ProducerStateManagerTest { partition, producerId, ProducerStateEntry.empty(producerId), - AppendOrigin.CLIENT + AppendOrigin.CLIENT, + stateManager.maybeCreateVerificationStateEntry(producerId, 0, producerEpoch) ) val firstOffsetMetadata = new LogOffsetMetadata(startOffset, segmentBaseOffset, 50 * relativeOffset) producerAppendInfo.appendDataBatch(producerEpoch, 0, 0, time.milliseconds(), @@ -1089,37 +1091,74 @@ class ProducerStateManagerTest { @Test def testEntryForVerification(): Unit = { - val originalEntry = stateManager.verificationStateEntry(producerId, true) + val originalEntry = stateManager.maybeCreateVerificationStateEntry(producerId, 0, 0) val originalEntryVerificationGuard = originalEntry.verificationGuard() def verifyEntry(producerId: Long, newEntry: VerificationStateEntry): Unit = { - val entry = stateManager.verificationStateEntry(producerId, false) + val entry = stateManager.verificationStateEntry(producerId) assertEquals(originalEntryVerificationGuard, entry.verificationGuard) assertEquals(entry.verificationGuard, newEntry.verificationGuard) } // If we already have an entry, reuse it. - val updatedEntry = stateManager.verificationStateEntry(producerId, true) + val updatedEntry = stateManager.maybeCreateVerificationStateEntry(producerId, 0, 0) verifyEntry(producerId, updatedEntry) // Add the transactional data and clear the entry. append(stateManager, producerId, 0, 0, offset = 0, isTransactional = true) stateManager.clearVerificationStateEntry(producerId) - assertNull(stateManager.verificationStateEntry(producerId, false)) + assertNull(stateManager.verificationStateEntry(producerId)) + } + + @Test + def testSequenceAndEpochInVerificationEntry(): Unit = { + val originalEntry = stateManager.maybeCreateVerificationStateEntry(producerId, 1, 0) + val originalEntryVerificationGuard = originalEntry.verificationGuard() + + def verifyEntry(producerId: Long, newEntry: VerificationStateEntry, expectedSequence: Int, expectedEpoch: Short): Unit = { + val entry = stateManager.verificationStateEntry(producerId) + assertEquals(originalEntryVerificationGuard, entry.verificationGuard) + assertEquals(entry.verificationGuard, newEntry.verificationGuard) + assertEquals(expectedSequence, entry.lowestSequence) + assertEquals(expectedEpoch, entry.epoch) + } + verifyEntry(producerId, originalEntry, 1, 0) + + // If we see a lower sequence, update to the lower one. + val updatedEntry = stateManager.maybeCreateVerificationStateEntry(producerId, 0, 0) + verifyEntry(producerId, updatedEntry, 0, 0) + + // If we see a new epoch that is higher, update the sequence. + val updatedEntryNewEpoch = stateManager.maybeCreateVerificationStateEntry(producerId, 2, 1) + verifyEntry(producerId, updatedEntryNewEpoch, 2, 1) + + // Ignore a lower epoch. + val updatedEntryOldEpoch = stateManager.maybeCreateVerificationStateEntry(producerId, 0, 0) + verifyEntry(producerId, updatedEntryOldEpoch, 2, 1) + } + + @Test + def testThrowOutOfOrderSequenceWithVerificationSequenceCheck(): Unit = { + val originalEntry = stateManager.maybeCreateVerificationStateEntry(producerId, 0, 0) + + // Trying to append with a higher sequence should fail + assertThrows(classOf[OutOfOrderSequenceException], () => append(stateManager, producerId, 0, 4, offset = 0, isTransactional = true)) + + assertEquals(originalEntry, stateManager.verificationStateEntry(producerId)) } @Test def testVerificationStateEntryExpiration(): Unit = { - val originalEntry = stateManager.verificationStateEntry(producerId, true) + val originalEntry = stateManager.maybeCreateVerificationStateEntry(producerId, 0, 0) // Before timeout we do not remove. Note: Accessing the verification entry does not update the time. time.sleep(producerStateManagerConfig.producerIdExpirationMs / 2) stateManager.removeExpiredProducers(time.milliseconds()) - assertEquals(originalEntry, stateManager.verificationStateEntry(producerId, false)) + assertEquals(originalEntry, stateManager.verificationStateEntry(producerId)) time.sleep((producerStateManagerConfig.producerIdExpirationMs / 2) + 1) stateManager.removeExpiredProducers(time.milliseconds()) - assertNull(stateManager.verificationStateEntry(producerId, false)) + assertNull(stateManager.verificationStateEntry(producerId)) } private def testLoadFromCorruptSnapshot(makeFileCorrupt: FileChannel => Unit): Unit = { diff --git a/core/src/test/scala/unit/kafka/log/UnifiedLogTest.scala b/core/src/test/scala/unit/kafka/log/UnifiedLogTest.scala index a1e22d8ba0f76..8c98bfb927899 100755 --- a/core/src/test/scala/unit/kafka/log/UnifiedLogTest.scala +++ b/core/src/test/scala/unit/kafka/log/UnifiedLogTest.scala @@ -3677,7 +3677,7 @@ class UnifiedLogTest { val logConfig = LogTestUtils.createLogConfig(segmentBytes = 2048 * 5) val log = createLog(logDir, logConfig, producerStateManagerConfig = producerStateManagerConfig) assertFalse(log.hasOngoingTransaction(producerId)) - assertNull(log.getOrMaybeCreateVerificationGuard(producerId)) + assertNull(log.verificationGuard(producerId)) val idempotentRecords = MemoryRecords.withIdempotentRecords( CompressionType.NONE, @@ -3688,15 +3688,6 @@ class UnifiedLogTest { new SimpleRecord("2".getBytes) ) - val verificationGuard = log.maybeStartTransactionVerification(producerId) - assertNotNull(verificationGuard) - - log.appendAsLeader(idempotentRecords, leaderEpoch = 0) - assertFalse(log.hasOngoingTransaction(producerId)) - - // Since we wrote idempotent records, we keep verification guard. - assertEquals(verificationGuard, log.getOrMaybeCreateVerificationGuard(producerId)) - val transactionalRecords = MemoryRecords.withTransactionalRecords( CompressionType.NONE, producerId, @@ -3706,13 +3697,23 @@ class UnifiedLogTest { new SimpleRecord("2".getBytes) ) + val verificationGuard = log.maybeStartTransactionVerification(producerId, sequence + 2, producerEpoch) + assertNotNull(verificationGuard) + + log.appendAsLeader(idempotentRecords, leaderEpoch = 0) + assertFalse(log.hasOngoingTransaction(producerId)) + + // Since we wrote idempotent records, we keep verification guard. + assertEquals(verificationGuard, log.verificationGuard(producerId)) + + // Now write the transactional records log.appendAsLeader(transactionalRecords, leaderEpoch = 0, verificationGuard = verificationGuard) assertTrue(log.hasOngoingTransaction(producerId)) // Verification guard should be cleared now. - assertNull(log.getOrMaybeCreateVerificationGuard(producerId)) + assertNull(log.verificationGuard(producerId)) // A subsequent maybeStartTransactionVerification will be empty since we are already verified. - assertNull(log.maybeStartTransactionVerification(producerId)) + assertNull(log.maybeStartTransactionVerification(producerId, sequence + 2, producerEpoch)) val endTransactionMarkerRecord = MemoryRecords.withEndTransactionMarker( producerId, @@ -3722,10 +3723,10 @@ class UnifiedLogTest { log.appendAsLeader(endTransactionMarkerRecord, origin = AppendOrigin.COORDINATOR, leaderEpoch = 0) assertFalse(log.hasOngoingTransaction(producerId)) - assertNull(log.getOrMaybeCreateVerificationGuard(producerId)) + assertNull(log.verificationGuard(producerId)) // A new maybeStartTransactionVerification will not be empty, as we need to verify the next transaction. - val newVerificationGuard = log.maybeStartTransactionVerification(producerId) + val newVerificationGuard = log.maybeStartTransactionVerification(producerId, sequence + 3, producerEpoch) assertNotNull(newVerificationGuard) assertNotEquals(verificationGuard, newVerificationGuard) } @@ -3739,7 +3740,7 @@ class UnifiedLogTest { val logConfig = LogTestUtils.createLogConfig(segmentBytes = 2048 * 5) val log = createLog(logDir, logConfig, producerStateManagerConfig = producerStateManagerConfig) - val verificationGuard = log.maybeStartTransactionVerification(producerId) + val verificationGuard = log.maybeStartTransactionVerification(producerId, 0, producerEpoch) assertNotNull(verificationGuard) val endTransactionMarkerRecord = MemoryRecords.withEndTransactionMarker( @@ -3750,7 +3751,7 @@ class UnifiedLogTest { log.appendAsLeader(endTransactionMarkerRecord, origin = AppendOrigin.COORDINATOR, leaderEpoch = 0) assertFalse(log.hasOngoingTransaction(producerId)) - assertNull(log.getOrMaybeCreateVerificationGuard(producerId)) + assertNull(log.verificationGuard(producerId)) } @Test @@ -3763,7 +3764,7 @@ class UnifiedLogTest { val logConfig = LogTestUtils.createLogConfig(segmentBytes = 2048 * 5) val log = createLog(logDir, logConfig, producerStateManagerConfig = producerStateManagerConfig) assertFalse(log.hasOngoingTransaction(producerId)) - assertNull(log.getOrMaybeCreateVerificationGuard(producerId)) + assertNull(log.verificationGuard(producerId)) val transactionalRecords = MemoryRecords.withTransactionalRecords( CompressionType.NONE, @@ -3774,7 +3775,7 @@ class UnifiedLogTest { new SimpleRecord("2".getBytes) ) - val verificationGuard = log.maybeStartTransactionVerification(producerId) + val verificationGuard = log.maybeStartTransactionVerification(producerId, sequence, producerEpoch) // Append should not throw error. log.appendAsLeader(transactionalRecords, leaderEpoch = 0, verificationGuard = verificationGuard) } diff --git a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala index 2bbda923529dc..abbcae1de61b4 100644 --- a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala +++ b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala @@ -2246,6 +2246,66 @@ class ReplicaManagerTest { } } + @Test + def testTransactionVerificationBlocksOutOfOrderSequence(): Unit = { + val tp0 = new TopicPartition(topic, 0) + val producerId = 24L + val producerEpoch = 0.toShort + val sequence = 6 + val node = new Node(0, "host1", 0) + val addPartitionsToTxnManager = mock(classOf[AddPartitionsToTxnManager]) + + val replicaManager = setUpReplicaManagerWithMockedAddPartitionsToTxnManager(addPartitionsToTxnManager, List(tp0), node) + try { + replicaManager.becomeLeaderOrFollower(1, + makeLeaderAndIsrRequest(topicIds(tp0.topic), tp0, Seq(0, 1), LeaderAndIsr(1, List(0, 1))), + (_, _) => ()) + + // Start with sequence 6 + val transactionalRecords = MemoryRecords.withTransactionalRecords(CompressionType.NONE, producerId, producerEpoch, sequence, + new SimpleRecord("message".getBytes)) + + val transactionToAdd = new AddPartitionsToTxnTransaction() + .setTransactionalId(transactionalId) + .setProducerId(producerId) + .setProducerEpoch(producerEpoch) + .setVerifyOnly(true) + .setTopics(new AddPartitionsToTxnTopicCollection( + Seq(new AddPartitionsToTxnTopic().setName(tp0.topic).setPartitions(Collections.singletonList(tp0.partition))).iterator.asJava + )) + + // We should add these partitions to the manager to verify. + val result = appendRecords(replicaManager, tp0, transactionalRecords, transactionalId = transactionalId, transactionStatePartition = Some(0)) + val appendCallback = ArgumentCaptor.forClass(classOf[AddPartitionsToTxnManager.AppendCallback]) + verify(addPartitionsToTxnManager, times(1)).addTxnData(ArgumentMatchers.eq(node), ArgumentMatchers.eq(transactionToAdd), appendCallback.capture()) + val verificationGuard = getVerificationGuard(replicaManager, tp0, producerId) + assertEquals(verificationGuard, getVerificationGuard(replicaManager, tp0, producerId)) + + // Confirm we did not write to the log and instead returned error. + val callback: AddPartitionsToTxnManager.AppendCallback = appendCallback.getValue() + callback(Map(tp0 -> Errors.NOT_COORDINATOR).toMap) + assertEquals(Errors.NOT_COORDINATOR, result.assertFired.error) + assertEquals(verificationGuard, getVerificationGuard(replicaManager, tp0, producerId)) + + // Try to append a higher sequence (7) after the first one failed with a retriable error. + val transactionalRecords2 = MemoryRecords.withTransactionalRecords(CompressionType.NONE, producerId, producerEpoch, sequence + 1, + new SimpleRecord("message".getBytes)) + + val result2 = appendRecords(replicaManager, tp0, transactionalRecords2, transactionalId = transactionalId, transactionStatePartition = Some(0)) + val appendCallback2 = ArgumentCaptor.forClass(classOf[AddPartitionsToTxnManager.AppendCallback]) + verify(addPartitionsToTxnManager, times(2)).addTxnData(ArgumentMatchers.eq(node), ArgumentMatchers.eq(transactionToAdd), appendCallback2.capture()) + assertEquals(verificationGuard, getVerificationGuard(replicaManager, tp0, producerId)) + + // Verification should succeed, but we expect to fail with OutOfOrderSequence and for the verification guard to remain. + val callback2: AddPartitionsToTxnManager.AppendCallback = appendCallback2.getValue() + callback2(Map.empty[TopicPartition, Errors].toMap) + assertEquals(verificationGuard, getVerificationGuard(replicaManager, tp0, producerId)) + assertEquals(Errors.OUT_OF_ORDER_SEQUENCE_NUMBER, result2.assertFired.error) + } finally { + replicaManager.shutdown(checkpointHW = false) + } + } + @Test def testTransactionVerificationGuardOnMultiplePartitions(): Unit = { val mockTimer = new MockTimer(time) @@ -2847,7 +2907,7 @@ class ReplicaManagerTest { private def getVerificationGuard(replicaManager: ReplicaManager, tp: TopicPartition, producerId: Long): Object = { - replicaManager.getPartitionOrException(tp).log.get.getOrMaybeCreateVerificationGuard(producerId) + replicaManager.getPartitionOrException(tp).log.get.verificationGuard(producerId) } private def setUpReplicaManagerWithMockedAddPartitionsToTxnManager(addPartitionsToTxnManager: AddPartitionsToTxnManager, diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/ProducerAppendInfo.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/ProducerAppendInfo.java index c800899641b16..bd2778287db32 100644 --- a/storage/src/main/java/org/apache/kafka/storage/internals/log/ProducerAppendInfo.java +++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/ProducerAppendInfo.java @@ -47,6 +47,7 @@ public class ProducerAppendInfo { private final long producerId; private final ProducerStateEntry currentEntry; private final AppendOrigin origin; + private final VerificationStateEntry verificationStateEntry; private final List transactions = new ArrayList<>(); private final ProducerStateEntry updatedEntry; @@ -54,25 +55,28 @@ public class ProducerAppendInfo { /** * Creates a new instance with the provided parameters. * - * @param topicPartition topic partition - * @param producerId The id of the producer appending to the log - * @param currentEntry The current entry associated with the producer id which contains metadata for a fixed number of - * the most recent appends made by the producer. Validation of the first incoming append will - * be made against the latest append in the current entry. New appends will replace older appends - * in the current entry so that the space overhead is constant. - * @param origin Indicates the origin of the append which implies the extent of validation. For example, offset - * commits, which originate from the group coordinator, do not have sequence numbers and therefore - * only producer epoch validation is done. Appends which come through replication are not validated - * (we assume the validation has already been done) and appends from clients require full validation. + * @param topicPartition topic partition + * @param producerId The id of the producer appending to the log + * @param currentEntry The current entry associated with the producer id which contains metadata for a fixed number of + * the most recent appends made by the producer. Validation of the first incoming append will + * be made against the latest append in the current entry. New appends will replace older appends + * in the current entry so that the space overhead is constant. + * @param origin Indicates the origin of the append which implies the extent of validation. For example, offset + * commits, which originate from the group coordinator, do not have sequence numbers and therefore + * only producer epoch validation is done. Appends which come through replication are not validated + * (we assume the validation has already been done) and appends from clients require full validation. + * @param verificationStateEntry The most recent entry used for verification if no append has been completed yet otherwise null */ public ProducerAppendInfo(TopicPartition topicPartition, long producerId, ProducerStateEntry currentEntry, - AppendOrigin origin) { + AppendOrigin origin, + VerificationStateEntry verificationStateEntry) { this.topicPartition = topicPartition; this.producerId = producerId; this.currentEntry = currentEntry; this.origin = origin; + this.verificationStateEntry = verificationStateEntry; updatedEntry = currentEntry.withProducerIdAndBatchMetadata(producerId, Optional.empty()); } @@ -105,6 +109,11 @@ private void checkProducerEpoch(short producerEpoch, long offset) { } private void checkSequence(short producerEpoch, int appendFirstSeq, long offset) { + if (verificationStateEntry != null && appendFirstSeq > verificationStateEntry.lowestSequence()) { + throw new OutOfOrderSequenceException("Out of order sequence number for producer " + producerId + " at " + + "offset " + offset + " in partition " + topicPartition + ": " + appendFirstSeq + + " (incoming seq. number), " + verificationStateEntry.lowestSequence() + " (earliest seen sequence)"); + } if (producerEpoch != updatedEntry.producerEpoch()) { if (appendFirstSeq != 0) { if (updatedEntry.producerEpoch() != RecordBatch.NO_PRODUCER_EPOCH) { diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/ProducerStateManager.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/ProducerStateManager.java index 6b88f73418943..da847982c1cf4 100644 --- a/storage/src/main/java/org/apache/kafka/storage/internals/log/ProducerStateManager.java +++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/ProducerStateManager.java @@ -187,16 +187,22 @@ private void clearProducerIds() { } /** - * Maybe create the VerificationStateEntry for a given producer ID. Return it if it exists, otherwise return null. + * Maybe create the VerificationStateEntry for a given producer ID and return it. + * This method also updates the sequence and epoch accordingly. */ - public VerificationStateEntry verificationStateEntry(long producerId, boolean createIfAbsent) { - return verificationStates.computeIfAbsent(producerId, pid -> { - if (createIfAbsent) - return new VerificationStateEntry(time.milliseconds()); - else { - return null; - } - }); + public VerificationStateEntry maybeCreateVerificationStateEntry(long producerId, int sequence, short epoch) { + VerificationStateEntry entry = verificationStates.computeIfAbsent(producerId, pid -> + new VerificationStateEntry(time.milliseconds(), sequence, epoch) + ); + entry.maybeUpdateLowestSequenceAndEpoch(sequence, epoch); + return entry; + } + + /** + * Return the VerificationStateEntry for the producer ID if it exists, otherwise return null. + */ + public VerificationStateEntry verificationStateEntry(long producerId) { + return verificationStates.get(producerId); } /** @@ -408,7 +414,7 @@ public void truncateAndReload(long logStartOffset, long logEndOffset, long curre public ProducerAppendInfo prepareUpdate(long producerId, AppendOrigin origin) { ProducerStateEntry currentEntry = lastEntry(producerId).orElse(ProducerStateEntry.empty(producerId)); - return new ProducerAppendInfo(topicPartition, producerId, currentEntry, origin); + return new ProducerAppendInfo(topicPartition, producerId, currentEntry, origin, verificationStateEntry(producerId)); } /** diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/VerificationStateEntry.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/VerificationStateEntry.java index 2464581fbe0b5..0a52095889aed 100644 --- a/storage/src/main/java/org/apache/kafka/storage/internals/log/VerificationStateEntry.java +++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/VerificationStateEntry.java @@ -22,16 +22,23 @@ * 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 verification guard object whenever we write data to the transaction or write an end marker for the transaction. + * + * We also store the lowest seen sequence to block a higher sequence from being written in the case of the lower sequence needing retries. + * * Any lingering entries that are never verified are removed via the producer state entry cleanup mechanism. */ public class VerificationStateEntry { final private long timestamp; final private Object verificationGuard; + private int lowestSequence; + private short epoch; - public VerificationStateEntry(long timestamp) { + public VerificationStateEntry(long timestamp, int sequence, short epoch) { this.timestamp = timestamp; this.verificationGuard = new Object(); + this.lowestSequence = sequence; + this.epoch = epoch; } public long timestamp() { @@ -41,4 +48,37 @@ public long timestamp() { public Object verificationGuard() { return verificationGuard; } + + public int lowestSequence() { + return lowestSequence; + } + + public short epoch() { + return epoch; + } + + /** + * An OutOfOrderSequence loop can happen for any idempotent/transactional producer when a lower sequence fails with + * a retriable error and a higher sequence is successfully written. The lower sequence will fail with + * OutOfOrderSequence and retry until retries run out. + * + * Here, we keep the lowest sequence seen in order to prevent an OutOfOrderSequence loop when verifying. This does + * not solve the error loop for idempotent producers or transactional producers that fail before verification + * starts. When verification fails with a retriable error (ie. NOT_COORDINATOR), the VerificationStateEntry + * maintains the lowest sequence number it sees and blocks higher sequences from being written to the log. However, + * if we encounter a new and lower sequence when verifying, we want to block sequences higher than that new + * sequence. Additionally, if the epoch is bumped, the sequence is reset and any previous sequence must be disregarded. + * + * Thus, we update the lowest sequence if there is a batch needing verification that has: + * a) a lower sequence for the same epoch + * b) a higher epoch -- update the epoch here too + */ + public void maybeUpdateLowestSequenceAndEpoch(int incomingSequence, short incomingEpoch) { + if (incomingEpoch == epoch && incomingSequence < this.lowestSequence) + this.lowestSequence = incomingSequence; + if (incomingEpoch > this.epoch) { + this.epoch = incomingEpoch; + this.lowestSequence = incomingSequence; + } + } }