You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2022/03/14 20:26:31 UTC
[GitHub] [kafka] hachikuji opened a new pull request #11891: KAFKA-13727; Preserve txn markers after partial segment cleaning
hachikuji opened a new pull request #11891:
URL: https://github.com/apache/kafka/pull/11891
It is possible to clean a segment partially if the offset map is filled. The highest offset that is reached becomes the new dirty offset after the cleaning completes. The data above this offset is nevertheless copied over to the new partially cleaned segment. Hence we need to ensure that the transaction index reflects aborted transactions in both the cleaned and uncleaned portion of the segment. Prior to this patch, this was not the case. We only collected the aborted transactions from the cleaned portion, which means that the reconstructed index would be incomplete. This can cause the aborted data to become effectively committed. It can also cause the deletion of the abort marker before the corresponding data has been removed (i.e. the aborted transaction becomes hanging).
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #11891: KAFKA-13727; Preserve txn markers after partial segment cleaning
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #11891:
URL: https://github.com/apache/kafka/pull/11891#discussion_r826494909
##########
File path: core/src/test/scala/unit/kafka/log/LogCleanerTest.scala
##########
@@ -270,6 +270,100 @@ class LogCleanerTest {
assertEquals(2L, logAppendInfo.lastOffset)
}
+ private def assertAllAbortedTxns(
+ expectedAbortedTxns: List[AbortedTxn],
+ log: UnifiedLog
+ ): Unit= {
+ val abortedTxns = log.collectAbortedTransactions(startOffset = 0L, upperBoundOffset = log.logEndOffset)
+ assertEquals(expectedAbortedTxns, abortedTxns)
+ }
+
+ private def assertAllTransactionsComplete(log: UnifiedLog): Unit = {
+ assertTrue(log.activeProducers.forall(_.currentTxnStartOffset() == -1))
+ }
+
+ @Test
+ def testMultiPassSegmentCleaningWithAbortedTransactions(): Unit = {
+ // Verify that the log cleaner preserves aborted transaction state (including the index)
+ // even if the cleaner cannot clean the whole segment in one pass.
+
+ val deleteRetentionMs = 50000
+ val offsetMapSlots = 4
+ val cleaner = makeCleaner(Int.MaxValue)
+ val logProps = new Properties()
+ logProps.put(LogConfig.DeleteRetentionMsProp, deleteRetentionMs.toString)
+ val log = makeLog(config = LogConfig.fromProps(logConfig.originals, logProps))
+
+ val producerEpoch = 0.toShort
+ val producerId1 = 1
+ val producerId2 = 2
+
+ val appendProducer1 = appendTransactionalAsLeader(log, producerId1, producerEpoch)
+ val appendProducer2 = appendTransactionalAsLeader(log, producerId2, producerEpoch)
+
+ appendProducer1(Seq(1, 2))
+ appendProducer2(Seq(2, 3))
+ appendProducer1(Seq(3, 4))
+ log.appendAsLeader(commitMarker(producerId1, producerEpoch), leaderEpoch = 0, origin = AppendOrigin.Replication)
+ log.appendAsLeader(commitMarker(producerId2, producerEpoch), leaderEpoch = 0, origin = AppendOrigin.Replication)
+ appendProducer1(Seq(2, 3))
+ log.appendAsLeader(abortMarker(producerId1, producerEpoch), leaderEpoch = 0, origin = AppendOrigin.Replication)
+ appendProducer2(Seq(4, 5))
+ appendProducer1(Seq(5, 6))
+ log.appendAsLeader(commitMarker(producerId1, producerEpoch), leaderEpoch = 0, origin = AppendOrigin.Replication)
+ log.appendAsLeader(abortMarker(producerId2, producerEpoch), leaderEpoch = 0, origin = AppendOrigin.Replication)
+ appendProducer2(Seq(6, 7))
+ log.appendAsLeader(commitMarker(producerId2, producerEpoch), leaderEpoch = 0, origin = AppendOrigin.Replication)
+ log.roll()
+
+ val expectedAbortedTxns = List(
+ new AbortedTxn(producerId=producerId1, firstOffset=8, lastOffset=10, lastStableOffset=11),
+ new AbortedTxn(producerId=producerId2, firstOffset=11, lastOffset=16, lastStableOffset=17)
+ )
+
+ assertAllTransactionsComplete(log)
+ assertAllAbortedTxns(expectedAbortedTxns, log)
+
+ var dirtyOffset = 0L
+ def cleanSegments(): Unit = {
+ val offsetMap = new FakeOffsetMap(slots = offsetMapSlots)
+ val segments = log.logSegments(0, log.activeSegment.baseOffset).toSeq
+ val stats = new CleanerStats(time)
+ cleaner.buildOffsetMap(log, dirtyOffset, log.activeSegment.baseOffset, offsetMap, stats)
+ cleaner.cleanSegments(log, segments, offsetMap, time.milliseconds(), stats, new CleanedTransactionMetadata, Long.MaxValue)
+ dirtyOffset = offsetMap.latestOffset + 1
+ }
+
+ // On the first pass, we should see the data from the aborted transactions deleted,
+ // but the markers should remain until the deletion retention time has passed.
+ cleanSegments()
+ assertEquals(4L, dirtyOffset)
+ assertEquals(List(0, 2, 4, 6, 7, 10, 13, 15, 16, 17, 19), batchBaseOffsetsInLog(log))
+ assertEquals(List(0, 2, 3, 4, 5, 6, 7, 10, 13, 14, 15, 16, 17, 18, 19), offsetsInLog(log))
+ assertAllTransactionsComplete(log)
+ assertAllAbortedTxns(expectedAbortedTxns, log)
+
+ // On the second pass, no data from the aborted transactions remains. The markers
+ // still cannot be removed from the log due to the retention time, but we do not
+ // need to record them in the transaction index since they are empty.
+ cleanSegments()
+ assertEquals(14, dirtyOffset)
+ assertEquals(List(0, 2, 4, 6, 7, 10, 13, 15, 16, 17, 19), batchBaseOffsetsInLog(log))
+ assertEquals(List(0, 2, 4, 5, 6, 7, 10, 13, 14, 15, 16, 17, 18, 19), offsetsInLog(log))
+ assertAllTransactionsComplete(log)
+ assertAllAbortedTxns(List(), log)
+
+ // On the last pass, wait for the retention time to expire. The abort markers
+ // (offsets 10 and 16) should be deleted.
+ time.sleep(deleteRetentionMs * 2)
+ cleanSegments()
+ assertEquals(20L, dirtyOffset)
+ assertEquals(List(0, 2, 4, 6, 7, 13, 15, 17, 19), batchBaseOffsetsInLog(log))
+ assertEquals(List(0, 2, 4, 5, 6, 7, 13, 15, 17, 18, 19), offsetsInLog(log))
Review comment:
I guess it's a little surprising, but the data from aborted transactions gets deleted even from the uncleaned portion of the log when we are building the new segment. In this test, that means the deletion horizon can be set in the second pass. I debating changing this behavior, but I couldn't come up with a strong reason to do so given that we have to do the work to build the transaction index anyway.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] junrao commented on a change in pull request #11891: KAFKA-13727; Preserve txn markers after partial segment cleaning
Posted by GitBox <gi...@apache.org>.
junrao commented on a change in pull request #11891:
URL: https://github.com/apache/kafka/pull/11891#discussion_r826482022
##########
File path: core/src/test/scala/unit/kafka/log/LogCleanerTest.scala
##########
@@ -270,6 +270,100 @@ class LogCleanerTest {
assertEquals(2L, logAppendInfo.lastOffset)
}
+ private def assertAllAbortedTxns(
+ expectedAbortedTxns: List[AbortedTxn],
+ log: UnifiedLog
+ ): Unit= {
+ val abortedTxns = log.collectAbortedTransactions(startOffset = 0L, upperBoundOffset = log.logEndOffset)
+ assertEquals(expectedAbortedTxns, abortedTxns)
+ }
+
+ private def assertAllTransactionsComplete(log: UnifiedLog): Unit = {
+ assertTrue(log.activeProducers.forall(_.currentTxnStartOffset() == -1))
+ }
+
+ @Test
+ def testMultiPassSegmentCleaningWithAbortedTransactions(): Unit = {
+ // Verify that the log cleaner preserves aborted transaction state (including the index)
+ // even if the cleaner cannot clean the whole segment in one pass.
+
+ val deleteRetentionMs = 50000
+ val offsetMapSlots = 4
+ val cleaner = makeCleaner(Int.MaxValue)
+ val logProps = new Properties()
+ logProps.put(LogConfig.DeleteRetentionMsProp, deleteRetentionMs.toString)
+ val log = makeLog(config = LogConfig.fromProps(logConfig.originals, logProps))
+
+ val producerEpoch = 0.toShort
+ val producerId1 = 1
+ val producerId2 = 2
+
+ val appendProducer1 = appendTransactionalAsLeader(log, producerId1, producerEpoch)
+ val appendProducer2 = appendTransactionalAsLeader(log, producerId2, producerEpoch)
+
+ appendProducer1(Seq(1, 2))
+ appendProducer2(Seq(2, 3))
+ appendProducer1(Seq(3, 4))
+ log.appendAsLeader(commitMarker(producerId1, producerEpoch), leaderEpoch = 0, origin = AppendOrigin.Replication)
+ log.appendAsLeader(commitMarker(producerId2, producerEpoch), leaderEpoch = 0, origin = AppendOrigin.Replication)
+ appendProducer1(Seq(2, 3))
+ log.appendAsLeader(abortMarker(producerId1, producerEpoch), leaderEpoch = 0, origin = AppendOrigin.Replication)
+ appendProducer2(Seq(4, 5))
+ appendProducer1(Seq(5, 6))
+ log.appendAsLeader(commitMarker(producerId1, producerEpoch), leaderEpoch = 0, origin = AppendOrigin.Replication)
+ log.appendAsLeader(abortMarker(producerId2, producerEpoch), leaderEpoch = 0, origin = AppendOrigin.Replication)
+ appendProducer2(Seq(6, 7))
+ log.appendAsLeader(commitMarker(producerId2, producerEpoch), leaderEpoch = 0, origin = AppendOrigin.Replication)
+ log.roll()
+
+ val expectedAbortedTxns = List(
+ new AbortedTxn(producerId=producerId1, firstOffset=8, lastOffset=10, lastStableOffset=11),
+ new AbortedTxn(producerId=producerId2, firstOffset=11, lastOffset=16, lastStableOffset=17)
+ )
+
+ assertAllTransactionsComplete(log)
+ assertAllAbortedTxns(expectedAbortedTxns, log)
+
+ var dirtyOffset = 0L
+ def cleanSegments(): Unit = {
+ val offsetMap = new FakeOffsetMap(slots = offsetMapSlots)
+ val segments = log.logSegments(0, log.activeSegment.baseOffset).toSeq
+ val stats = new CleanerStats(time)
+ cleaner.buildOffsetMap(log, dirtyOffset, log.activeSegment.baseOffset, offsetMap, stats)
+ cleaner.cleanSegments(log, segments, offsetMap, time.milliseconds(), stats, new CleanedTransactionMetadata, Long.MaxValue)
+ dirtyOffset = offsetMap.latestOffset + 1
+ }
+
+ // On the first pass, we should see the data from the aborted transactions deleted,
+ // but the markers should remain until the deletion retention time has passed.
+ cleanSegments()
+ assertEquals(4L, dirtyOffset)
+ assertEquals(List(0, 2, 4, 6, 7, 10, 13, 15, 16, 17, 19), batchBaseOffsetsInLog(log))
+ assertEquals(List(0, 2, 3, 4, 5, 6, 7, 10, 13, 14, 15, 16, 17, 18, 19), offsetsInLog(log))
+ assertAllTransactionsComplete(log)
+ assertAllAbortedTxns(expectedAbortedTxns, log)
+
+ // On the second pass, no data from the aborted transactions remains. The markers
+ // still cannot be removed from the log due to the retention time, but we do not
+ // need to record them in the transaction index since they are empty.
+ cleanSegments()
+ assertEquals(14, dirtyOffset)
+ assertEquals(List(0, 2, 4, 6, 7, 10, 13, 15, 16, 17, 19), batchBaseOffsetsInLog(log))
+ assertEquals(List(0, 2, 4, 5, 6, 7, 10, 13, 14, 15, 16, 17, 18, 19), offsetsInLog(log))
+ assertAllTransactionsComplete(log)
+ assertAllAbortedTxns(List(), log)
+
+ // On the last pass, wait for the retention time to expire. The abort markers
+ // (offsets 10 and 16) should be deleted.
+ time.sleep(deleteRetentionMs * 2)
+ cleanSegments()
+ assertEquals(20L, dirtyOffset)
+ assertEquals(List(0, 2, 4, 6, 7, 13, 15, 17, 19), batchBaseOffsetsInLog(log))
+ assertEquals(List(0, 2, 4, 5, 6, 7, 13, 15, 17, 18, 19), offsetsInLog(log))
Review comment:
I am wondering why offset 16 (abort for p2) is removed in this round of cleaning. This is the first time that we clean offset 16. I was expecting this round only sets the delete horizon for the abort marker.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] junrao commented on a change in pull request #11891: KAFKA-13727; Preserve txn markers after partial segment cleaning
Posted by GitBox <gi...@apache.org>.
junrao commented on a change in pull request #11891:
URL: https://github.com/apache/kafka/pull/11891#discussion_r827250663
##########
File path: core/src/test/scala/unit/kafka/log/LogCleanerTest.scala
##########
@@ -270,6 +270,100 @@ class LogCleanerTest {
assertEquals(2L, logAppendInfo.lastOffset)
}
+ private def assertAllAbortedTxns(
+ expectedAbortedTxns: List[AbortedTxn],
+ log: UnifiedLog
+ ): Unit= {
+ val abortedTxns = log.collectAbortedTransactions(startOffset = 0L, upperBoundOffset = log.logEndOffset)
+ assertEquals(expectedAbortedTxns, abortedTxns)
+ }
+
+ private def assertAllTransactionsComplete(log: UnifiedLog): Unit = {
+ assertTrue(log.activeProducers.forall(_.currentTxnStartOffset() == -1))
+ }
+
+ @Test
+ def testMultiPassSegmentCleaningWithAbortedTransactions(): Unit = {
+ // Verify that the log cleaner preserves aborted transaction state (including the index)
+ // even if the cleaner cannot clean the whole segment in one pass.
+
+ val deleteRetentionMs = 50000
+ val offsetMapSlots = 4
+ val cleaner = makeCleaner(Int.MaxValue)
+ val logProps = new Properties()
+ logProps.put(LogConfig.DeleteRetentionMsProp, deleteRetentionMs.toString)
+ val log = makeLog(config = LogConfig.fromProps(logConfig.originals, logProps))
+
+ val producerEpoch = 0.toShort
+ val producerId1 = 1
+ val producerId2 = 2
+
+ val appendProducer1 = appendTransactionalAsLeader(log, producerId1, producerEpoch)
+ val appendProducer2 = appendTransactionalAsLeader(log, producerId2, producerEpoch)
+
+ appendProducer1(Seq(1, 2))
+ appendProducer2(Seq(2, 3))
+ appendProducer1(Seq(3, 4))
+ log.appendAsLeader(commitMarker(producerId1, producerEpoch), leaderEpoch = 0, origin = AppendOrigin.Replication)
+ log.appendAsLeader(commitMarker(producerId2, producerEpoch), leaderEpoch = 0, origin = AppendOrigin.Replication)
+ appendProducer1(Seq(2, 3))
+ log.appendAsLeader(abortMarker(producerId1, producerEpoch), leaderEpoch = 0, origin = AppendOrigin.Replication)
+ appendProducer2(Seq(4, 5))
+ appendProducer1(Seq(5, 6))
+ log.appendAsLeader(commitMarker(producerId1, producerEpoch), leaderEpoch = 0, origin = AppendOrigin.Replication)
+ log.appendAsLeader(abortMarker(producerId2, producerEpoch), leaderEpoch = 0, origin = AppendOrigin.Replication)
+ appendProducer2(Seq(6, 7))
+ log.appendAsLeader(commitMarker(producerId2, producerEpoch), leaderEpoch = 0, origin = AppendOrigin.Replication)
+ log.roll()
+
+ val expectedAbortedTxns = List(
+ new AbortedTxn(producerId=producerId1, firstOffset=8, lastOffset=10, lastStableOffset=11),
+ new AbortedTxn(producerId=producerId2, firstOffset=11, lastOffset=16, lastStableOffset=17)
+ )
+
+ assertAllTransactionsComplete(log)
+ assertAllAbortedTxns(expectedAbortedTxns, log)
+
+ var dirtyOffset = 0L
+ def cleanSegments(): Unit = {
+ val offsetMap = new FakeOffsetMap(slots = offsetMapSlots)
+ val segments = log.logSegments(0, log.activeSegment.baseOffset).toSeq
+ val stats = new CleanerStats(time)
+ cleaner.buildOffsetMap(log, dirtyOffset, log.activeSegment.baseOffset, offsetMap, stats)
+ cleaner.cleanSegments(log, segments, offsetMap, time.milliseconds(), stats, new CleanedTransactionMetadata, Long.MaxValue)
+ dirtyOffset = offsetMap.latestOffset + 1
+ }
+
+ // On the first pass, we should see the data from the aborted transactions deleted,
+ // but the markers should remain until the deletion retention time has passed.
+ cleanSegments()
+ assertEquals(4L, dirtyOffset)
+ assertEquals(List(0, 2, 4, 6, 7, 10, 13, 15, 16, 17, 19), batchBaseOffsetsInLog(log))
+ assertEquals(List(0, 2, 3, 4, 5, 6, 7, 10, 13, 14, 15, 16, 17, 18, 19), offsetsInLog(log))
+ assertAllTransactionsComplete(log)
+ assertAllAbortedTxns(expectedAbortedTxns, log)
+
+ // On the second pass, no data from the aborted transactions remains. The markers
+ // still cannot be removed from the log due to the retention time, but we do not
+ // need to record them in the transaction index since they are empty.
+ cleanSegments()
+ assertEquals(14, dirtyOffset)
+ assertEquals(List(0, 2, 4, 6, 7, 10, 13, 15, 16, 17, 19), batchBaseOffsetsInLog(log))
+ assertEquals(List(0, 2, 4, 5, 6, 7, 10, 13, 14, 15, 16, 17, 18, 19), offsetsInLog(log))
+ assertAllTransactionsComplete(log)
+ assertAllAbortedTxns(List(), log)
+
+ // On the last pass, wait for the retention time to expire. The abort markers
+ // (offsets 10 and 16) should be deleted.
+ time.sleep(deleteRetentionMs * 2)
+ cleanSegments()
+ assertEquals(20L, dirtyOffset)
+ assertEquals(List(0, 2, 4, 6, 7, 13, 15, 17, 19), batchBaseOffsetsInLog(log))
+ assertEquals(List(0, 2, 4, 5, 6, 7, 13, 15, 17, 18, 19), offsetsInLog(log))
Review comment:
Thanks for the explanation. Make sense to me.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji merged pull request #11891: KAFKA-13727; Preserve txn markers after partial segment cleaning
Posted by GitBox <gi...@apache.org>.
hachikuji merged pull request #11891:
URL: https://github.com/apache/kafka/pull/11891
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #11891: KAFKA-13727; Preserve txn markers after partial segment cleaning
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #11891:
URL: https://github.com/apache/kafka/pull/11891#discussion_r826359121
##########
File path: clients/src/main/java/org/apache/kafka/common/record/RecordBatchIterator.java
##########
@@ -36,6 +38,8 @@ protected T makeNext() {
if (batch == null)
return allDone();
return batch;
+ } catch (EOFException e) {
Review comment:
`LogCleanerTest.testCorruptMessageSizeLargerThanBytesAvailable` was failing after the change to `readNextOffset` in `LogCleaner` due to an unexpected EOF. It seems reasonable to catch it here and raise as a `CorruptRecordException` (which is also a `KafkaException`). This made the test pass.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org