From d162c26242a1fa25479b9067e8238dbdaaf4c75a Mon Sep 17 00:00:00 2001 From: Jason Gustafson Date: Fri, 3 Apr 2020 10:19:55 -0700 Subject: [PATCH 1/2] KAFKA-9807; Protect LSO reads from concurrent high-watermark updates --- core/src/main/scala/kafka/log/Log.scala | 7 ++- .../test/scala/unit/kafka/log/LogTest.scala | 56 +++++++++++++++++++ 2 files changed, 61 insertions(+), 2 deletions(-) diff --git a/core/src/main/scala/kafka/log/Log.scala b/core/src/main/scala/kafka/log/Log.scala index e9252d640c609..0f756bbed646f 100644 --- a/core/src/main/scala/kafka/log/Log.scala +++ b/core/src/main/scala/kafka/log/Log.scala @@ -420,8 +420,11 @@ class Log(@volatile private var _dir: File, private def fetchLastStableOffsetMetadata: LogOffsetMetadata = { checkIfMemoryMappedBufferClosed() + // cache the current high watermark to avoid a concurrent update invalidating the range check + val highWatermarkMetadata = fetchHighWatermarkMetadata + firstUnstableOffsetMetadata match { - case Some(offsetMetadata) if offsetMetadata.messageOffset < highWatermark => + case Some(offsetMetadata) if offsetMetadata.messageOffset < highWatermarkMetadata.messageOffset => if (offsetMetadata.messageOffsetOnly) { lock synchronized { val fullOffset = convertToOffsetMetadataOrThrow(offsetMetadata.messageOffset) @@ -432,7 +435,7 @@ class Log(@volatile private var _dir: File, } else { offsetMetadata } - case _ => fetchHighWatermarkMetadata + case _ => highWatermarkMetadata } } diff --git a/core/src/test/scala/unit/kafka/log/LogTest.scala b/core/src/test/scala/unit/kafka/log/LogTest.scala index 762edd2090095..1245081aff4ac 100755 --- a/core/src/test/scala/unit/kafka/log/LogTest.scala +++ b/core/src/test/scala/unit/kafka/log/LogTest.scala @@ -20,6 +20,7 @@ package kafka.log import java.io._ import java.nio.ByteBuffer import java.nio.file.{Files, Paths} +import java.util.concurrent.{Callable, Executors} import java.util.regex.Pattern import java.util.{Collections, Optional, Properties} @@ -3649,6 +3650,61 @@ class LogTest { assertEquals(None, log.firstUnstableOffset) } + @Test + def testReadCommittedWithConcurrentHighWatermarkUpdates(): Unit = { + val logConfig = LogTest.createLogConfig(segmentBytes = 1024 * 1024 * 5) + val log = createLog(logDir, logConfig) + val lastOffset = 50L + + val producerEpoch = 0.toShort + val producerId = 15L + val appendProducer = appendTransactionalAsLeader(log, producerId, producerEpoch) + + // Thread 1 writes single-record transactions and attempts to read them + // before they have been aborted, and then aborts them + val txnVerifier = new Callable[Int]() { + override def call(): Int = { + var nonEmptyReads = 0 + while (log.logEndOffset < lastOffset) { + val currentLogEndOffset = log.logEndOffset + + appendProducer(1) + + val readInfo = log.read( + startOffset = currentLogEndOffset, + maxLength = Int.MaxValue, + isolation = FetchTxnCommitted, + minOneMessage = false) + + if (readInfo.records.sizeInBytes() > 0) + nonEmptyReads += 1 + + appendEndTxnMarkerAsLeader(log, producerId, producerEpoch, ControlRecordType.ABORT) + } + nonEmptyReads + } + } + + // Thread 2 watches the log and updates the high watermark + val hwUpdater = new Runnable() { + override def run(): Unit = { + while (log.logEndOffset < lastOffset) { + log.updateHighWatermark(log.logEndOffset) + } + } + } + + val executor = Executors.newFixedThreadPool(2) + executor.submit(hwUpdater) + + val future = executor.submit(txnVerifier) + val nonEmptyReads = future.get() + + assertEquals(0, nonEmptyReads) + + executor.shutdownNow() + } + @Test def testTransactionIndexUpdated(): Unit = { val logConfig = LogTest.createLogConfig(segmentBytes = 1024 * 1024 * 5) From ef0058146efe1bbd9b848be3395c4be571b4842f Mon Sep 17 00:00:00 2001 From: Jason Gustafson Date: Fri, 3 Apr 2020 10:54:30 -0700 Subject: [PATCH 2/2] Address feedback on test case --- .../test/scala/unit/kafka/log/LogTest.scala | 52 +++++++++---------- 1 file changed, 25 insertions(+), 27 deletions(-) diff --git a/core/src/test/scala/unit/kafka/log/LogTest.scala b/core/src/test/scala/unit/kafka/log/LogTest.scala index 1245081aff4ac..0cc19dda9e712 100755 --- a/core/src/test/scala/unit/kafka/log/LogTest.scala +++ b/core/src/test/scala/unit/kafka/log/LogTest.scala @@ -3662,47 +3662,45 @@ class LogTest { // Thread 1 writes single-record transactions and attempts to read them // before they have been aborted, and then aborts them - val txnVerifier = new Callable[Int]() { - override def call(): Int = { - var nonEmptyReads = 0 - while (log.logEndOffset < lastOffset) { - val currentLogEndOffset = log.logEndOffset + val txnWriteAndReadLoop: Callable[Int] = () => { + var nonEmptyReads = 0 + while (log.logEndOffset < lastOffset) { + val currentLogEndOffset = log.logEndOffset - appendProducer(1) + appendProducer(1) - val readInfo = log.read( - startOffset = currentLogEndOffset, - maxLength = Int.MaxValue, - isolation = FetchTxnCommitted, - minOneMessage = false) + val readInfo = log.read( + startOffset = currentLogEndOffset, + maxLength = Int.MaxValue, + isolation = FetchTxnCommitted, + minOneMessage = false) - if (readInfo.records.sizeInBytes() > 0) - nonEmptyReads += 1 + if (readInfo.records.sizeInBytes() > 0) + nonEmptyReads += 1 - appendEndTxnMarkerAsLeader(log, producerId, producerEpoch, ControlRecordType.ABORT) - } - nonEmptyReads + appendEndTxnMarkerAsLeader(log, producerId, producerEpoch, ControlRecordType.ABORT) } + nonEmptyReads } // Thread 2 watches the log and updates the high watermark - val hwUpdater = new Runnable() { - override def run(): Unit = { - while (log.logEndOffset < lastOffset) { - log.updateHighWatermark(log.logEndOffset) - } + val hwUpdateLoop: Runnable = () => { + while (log.logEndOffset < lastOffset) { + log.updateHighWatermark(log.logEndOffset) } } val executor = Executors.newFixedThreadPool(2) - executor.submit(hwUpdater) - - val future = executor.submit(txnVerifier) - val nonEmptyReads = future.get() + try { + executor.submit(hwUpdateLoop) - assertEquals(0, nonEmptyReads) + val future = executor.submit(txnWriteAndReadLoop) + val nonEmptyReads = future.get() - executor.shutdownNow() + assertEquals(0, nonEmptyReads) + } finally { + executor.shutdownNow() + } } @Test