From f0466b2420dc500112b4839678564996c0677d2e Mon Sep 17 00:00:00 2001 From: Kowshik Prakasam Date: Tue, 8 Dec 2020 22:28:00 -0800 Subject: [PATCH 1/4] MINOR: Use the correct ProducerStateManager when updating producers --- core/src/main/scala/kafka/log/Log.scala | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/core/src/main/scala/kafka/log/Log.scala b/core/src/main/scala/kafka/log/Log.scala index 9677dc4f6c7a7..1b0df21a67fcc 100644 --- a/core/src/main/scala/kafka/log/Log.scala +++ b/core/src/main/scala/kafka/log/Log.scala @@ -953,7 +953,9 @@ class Log(@volatile private var _dir: File, val completedTxns = ListBuffer.empty[CompletedTxn] records.batches.forEach { batch => if (batch.hasProducerId) { - val maybeCompletedTxn = updateProducers(batch, + val maybeCompletedTxn = updateProducers( + producerStateManager, + batch, loadedProducers, firstOffsetMetadata = None, origin = AppendOrigin.Replication) @@ -1349,7 +1351,7 @@ class Log(@volatile private var _dir: File, else None - val maybeCompletedTxn = updateProducers(batch, updatedProducers, firstOffsetMetadata, origin) + val maybeCompletedTxn = updateProducers(producerStateManager, batch, updatedProducers, firstOffsetMetadata, origin) maybeCompletedTxn.foreach(completedTxns += _) } @@ -1456,7 +1458,8 @@ class Log(@volatile private var _dir: File, RecordConversionStats.EMPTY, sourceCodec, targetCodec, shallowMessageCount, validBytesCount, monotonic, lastOffsetOfFirstBatch) } - private def updateProducers(batch: RecordBatch, + private def updateProducers(producerStateManager: ProducerStateManager, + batch: RecordBatch, producers: mutable.Map[Long, ProducerAppendInfo], firstOffsetMetadata: Option[LogOffsetMetadata], origin: AppendOrigin): Option[CompletedTxn] = { From 6d348bcc12f9e8b2ddf0552c089945b4eda67538 Mon Sep 17 00:00:00 2001 From: Kowshik Prakasam Date: Thu, 10 Dec 2020 13:11:32 -0800 Subject: [PATCH 2/4] Address comments from @ijuma --- core/src/main/scala/kafka/log/Log.scala | 210 ++++++++++++------------ 1 file changed, 108 insertions(+), 102 deletions(-) diff --git a/core/src/main/scala/kafka/log/Log.scala b/core/src/main/scala/kafka/log/Log.scala index 1b0df21a67fcc..2734cf52802dd 100644 --- a/core/src/main/scala/kafka/log/Log.scala +++ b/core/src/main/scala/kafka/log/Log.scala @@ -864,83 +864,11 @@ class Log(@volatile private var _dir: File, recoveryPoint } - // Rebuild producer state until lastOffset. This method may be called from the recovery code path, and thus must be - // free of all side-effects, i.e. it must not update any log-specific state. private def rebuildProducerState(lastOffset: Long, reloadFromCleanShutdown: Boolean, producerStateManager: ProducerStateManager): Unit = lock synchronized { - checkIfMemoryMappedBufferClosed() - val messageFormatVersion = config.messageFormatVersion.recordVersion.value - val segments = logSegments - val offsetsToSnapshot = - if (segments.nonEmpty) { - val nextLatestSegmentBaseOffset = lowerSegment(segments.last.baseOffset).map(_.baseOffset) - Seq(nextLatestSegmentBaseOffset, Some(segments.last.baseOffset), Some(lastOffset)) - } else { - Seq(Some(lastOffset)) - } - info(s"Loading producer state till offset $lastOffset with message format version $messageFormatVersion") - - // We want to avoid unnecessary scanning of the log to build the producer state when the broker is being - // upgraded. The basic idea is to use the absence of producer snapshot files to detect the upgrade case, - // but we have to be careful not to assume too much in the presence of broker failures. The two most common - // upgrade cases in which we expect to find no snapshots are the following: - // - // 1. The broker has been upgraded, but the topic is still on the old message format. - // 2. The broker has been upgraded, the topic is on the new message format, and we had a clean shutdown. - // - // If we hit either of these cases, we skip producer state loading and write a new snapshot at the log end - // offset (see below). The next time the log is reloaded, we will load producer state using this snapshot - // (or later snapshots). Otherwise, if there is no snapshot file, then we have to rebuild producer state - // from the first segment. - if (messageFormatVersion < RecordBatch.MAGIC_VALUE_V2 || - (producerStateManager.latestSnapshotOffset.isEmpty && reloadFromCleanShutdown)) { - // To avoid an expensive scan through all of the segments, we take empty snapshots from the start of the - // last two segments and the last offset. This should avoid the full scan in the case that the log needs - // truncation. - offsetsToSnapshot.flatten.foreach { offset => - producerStateManager.updateMapEndOffset(offset) - producerStateManager.takeSnapshot() - } - } else { - val isEmptyBeforeTruncation = producerStateManager.isEmpty && producerStateManager.mapEndOffset >= lastOffset - producerStateManager.truncateAndReload(logStartOffset, lastOffset, time.milliseconds()) - - // Only do the potentially expensive reloading if the last snapshot offset is lower than the log end - // offset (which would be the case on first startup) and there were active producers prior to truncation - // (which could be the case if truncating after initial loading). If there weren't, then truncating - // shouldn't change that fact (although it could cause a producerId to expire earlier than expected), - // and we can skip the loading. This is an optimization for users which are not yet using - // idempotent/transactional features yet. - if (lastOffset > producerStateManager.mapEndOffset && !isEmptyBeforeTruncation) { - val segmentOfLastOffset = floorLogSegment(lastOffset) - - logSegments(producerStateManager.mapEndOffset, lastOffset).foreach { segment => - val startOffset = Utils.max(segment.baseOffset, producerStateManager.mapEndOffset, logStartOffset) - producerStateManager.updateMapEndOffset(startOffset) - - if (offsetsToSnapshot.contains(Some(segment.baseOffset))) - producerStateManager.takeSnapshot() - - val maxPosition = if (segmentOfLastOffset.contains(segment)) { - Option(segment.translateOffset(lastOffset)) - .map(_.position) - .getOrElse(segment.size) - } else { - segment.size - } - - val fetchDataInfo = segment.read(startOffset, - maxSize = Int.MaxValue, - maxPosition = maxPosition, - minOneMessage = false) - if (fetchDataInfo != null) - loadProducersFromLog(producerStateManager, fetchDataInfo.records) - } - } - producerStateManager.updateMapEndOffset(lastOffset) - producerStateManager.takeSnapshot() - } + info(s"Loading producer state till offset $lastOffset with message format version ${config.messageFormatVersion.recordVersion.value}") + Log.rebuildProducerState(this, lastOffset, reloadFromCleanShutdown, producerStateManager) } private def loadProducerState(lastOffset: Long, reloadFromCleanShutdown: Boolean): Unit = lock synchronized { @@ -948,24 +876,6 @@ class Log(@volatile private var _dir: File, maybeIncrementFirstUnstableOffset() } - private def loadProducersFromLog(producerStateManager: ProducerStateManager, records: Records): Unit = { - val loadedProducers = mutable.Map.empty[Long, ProducerAppendInfo] - val completedTxns = ListBuffer.empty[CompletedTxn] - records.batches.forEach { batch => - if (batch.hasProducerId) { - val maybeCompletedTxn = updateProducers( - producerStateManager, - batch, - loadedProducers, - firstOffsetMetadata = None, - origin = AppendOrigin.Replication) - maybeCompletedTxn.foreach(completedTxns += _) - } - } - loadedProducers.values.foreach(producerStateManager.update) - completedTxns.foreach(producerStateManager.completeTxn) - } - private[log] def activeProducersWithLastSequence: Map[Long, Int] = lock synchronized { producerStateManager.activeProducers.map { case (producerId, producerIdEntry) => (producerId, producerIdEntry.lastSeq) @@ -1458,16 +1368,6 @@ class Log(@volatile private var _dir: File, RecordConversionStats.EMPTY, sourceCodec, targetCodec, shallowMessageCount, validBytesCount, monotonic, lastOffsetOfFirstBatch) } - private def updateProducers(producerStateManager: ProducerStateManager, - batch: RecordBatch, - producers: mutable.Map[Long, ProducerAppendInfo], - firstOffsetMetadata: Option[LogOffsetMetadata], - origin: AppendOrigin): Option[CompletedTxn] = { - val producerId = batch.producerId - val appendInfo = producers.getOrElseUpdate(producerId, producerStateManager.prepareUpdate(producerId, origin)) - appendInfo.append(batch, firstOffsetMetadata) - } - /** * Trim any invalid bytes from the end of this message set (if there are any) * @@ -2673,6 +2573,112 @@ object Log { private def isLogFile(file: File): Boolean = file.getPath.endsWith(LogFileSuffix) + // Rebuild producer state until lastOffset. This method may be called from the recovery code path, and thus must be + // free of all side-effects, i.e. it must not update any log-specific state. + private def rebuildProducerState(log: Log, + lastOffset: Long, + reloadFromCleanShutdown: Boolean, + producerStateManager: ProducerStateManager): Unit = { + val messageFormatVersion = log.config.messageFormatVersion.recordVersion.value + val segments = log.logSegments + val offsetsToSnapshot = + if (segments.nonEmpty) { + val nextLatestSegmentBaseOffset = log.lowerSegment(segments.last.baseOffset).map(_.baseOffset) + Seq(nextLatestSegmentBaseOffset, Some(segments.last.baseOffset), Some(lastOffset)) + } else { + Seq(Some(lastOffset)) + } + + // We want to avoid unnecessary scanning of the log to build the producer state when the broker is being + // upgraded. The basic idea is to use the absence of producer snapshot files to detect the upgrade case, + // but we have to be careful not to assume too much in the presence of broker failures. The two most common + // upgrade cases in which we expect to find no snapshots are the following: + // + // 1. The broker has been upgraded, but the topic is still on the old message format. + // 2. The broker has been upgraded, the topic is on the new message format, and we had a clean shutdown. + // + // If we hit either of these cases, we skip producer state loading and write a new snapshot at the log end + // offset (see below). The next time the log is reloaded, we will load producer state using this snapshot + // (or later snapshots). Otherwise, if there is no snapshot file, then we have to rebuild producer state + // from the first segment. + if (messageFormatVersion < RecordBatch.MAGIC_VALUE_V2 || + (producerStateManager.latestSnapshotOffset.isEmpty && reloadFromCleanShutdown)) { + // To avoid an expensive scan through all of the segments, we take empty snapshots from the start of the + // last two segments and the last offset. This should avoid the full scan in the case that the log needs + // truncation. + offsetsToSnapshot.flatten.foreach { offset => + producerStateManager.updateMapEndOffset(offset) + producerStateManager.takeSnapshot() + } + } else { + val isEmptyBeforeTruncation = producerStateManager.isEmpty && producerStateManager.mapEndOffset >= lastOffset + producerStateManager.truncateAndReload(log.logStartOffset, lastOffset, log.time.milliseconds()) + + // Only do the potentially expensive reloading if the last snapshot offset is lower than the log end + // offset (which would be the case on first startup) and there were active producers prior to truncation + // (which could be the case if truncating after initial loading). If there weren't, then truncating + // shouldn't change that fact (although it could cause a producerId to expire earlier than expected), + // and we can skip the loading. This is an optimization for users which are not yet using + // idempotent/transactional features yet. + if (lastOffset > producerStateManager.mapEndOffset && !isEmptyBeforeTruncation) { + val segmentOfLastOffset = log.floorLogSegment(lastOffset) + + log.logSegments(producerStateManager.mapEndOffset, lastOffset).foreach { segment => + val startOffset = Utils.max(segment.baseOffset, producerStateManager.mapEndOffset, log.logStartOffset) + producerStateManager.updateMapEndOffset(startOffset) + + if (offsetsToSnapshot.contains(Some(segment.baseOffset))) + producerStateManager.takeSnapshot() + + val maxPosition = if (segmentOfLastOffset.contains(segment)) { + Option(segment.translateOffset(lastOffset)) + .map(_.position) + .getOrElse(segment.size) + } else { + segment.size + } + + val fetchDataInfo = segment.read(startOffset, + maxSize = Int.MaxValue, + maxPosition = maxPosition, + minOneMessage = false) + if (fetchDataInfo != null) + loadProducersFromLog(producerStateManager, fetchDataInfo.records) + } + } + producerStateManager.updateMapEndOffset(lastOffset) + producerStateManager.takeSnapshot() + } + } + + private def loadProducersFromLog(producerStateManager: ProducerStateManager, records: Records): Unit = { + val loadedProducers = mutable.Map.empty[Long, ProducerAppendInfo] + val completedTxns = ListBuffer.empty[CompletedTxn] + records.batches.forEach { batch => + if (batch.hasProducerId) { + val maybeCompletedTxn = updateProducers( + producerStateManager, + batch, + loadedProducers, + firstOffsetMetadata = None, + origin = AppendOrigin.Replication) + maybeCompletedTxn.foreach(completedTxns += _) + } + } + loadedProducers.values.foreach(producerStateManager.update) + completedTxns.foreach(producerStateManager.completeTxn) + } + + private def updateProducers(producerStateManager: ProducerStateManager, + batch: RecordBatch, + producers: mutable.Map[Long, ProducerAppendInfo], + firstOffsetMetadata: Option[LogOffsetMetadata], + origin: AppendOrigin): Option[CompletedTxn] = { + val producerId = batch.producerId + val appendInfo = producers.getOrElseUpdate(producerId, producerStateManager.prepareUpdate(producerId, origin)) + appendInfo.append(batch, firstOffsetMetadata) + } + } object LogMetricNames { From 5d798b7d884bd11e43884318ae3fc8d12e635c40 Mon Sep 17 00:00:00 2001 From: Kowshik Prakasam Date: Thu, 10 Dec 2020 18:13:59 -0800 Subject: [PATCH 3/4] Address comments from @junrao --- core/src/main/scala/kafka/log/Log.scala | 154 ++++++++++++------------ 1 file changed, 74 insertions(+), 80 deletions(-) diff --git a/core/src/main/scala/kafka/log/Log.scala b/core/src/main/scala/kafka/log/Log.scala index 2734cf52802dd..38b279678dceb 100644 --- a/core/src/main/scala/kafka/log/Log.scala +++ b/core/src/main/scala/kafka/log/Log.scala @@ -864,11 +864,83 @@ class Log(@volatile private var _dir: File, recoveryPoint } + // Rebuild producer state until lastOffset. This method may be called from the recovery code path, and thus must be + // free of all side-effects, i.e. it must not update any log-specific state. private def rebuildProducerState(lastOffset: Long, reloadFromCleanShutdown: Boolean, producerStateManager: ProducerStateManager): Unit = lock synchronized { - info(s"Loading producer state till offset $lastOffset with message format version ${config.messageFormatVersion.recordVersion.value}") - Log.rebuildProducerState(this, lastOffset, reloadFromCleanShutdown, producerStateManager) + checkIfMemoryMappedBufferClosed() + val messageFormatVersion = config.messageFormatVersion.recordVersion.value + val segments = logSegments + val offsetsToSnapshot = + if (segments.nonEmpty) { + val nextLatestSegmentBaseOffset = lowerSegment(segments.last.baseOffset).map(_.baseOffset) + Seq(nextLatestSegmentBaseOffset, Some(segments.last.baseOffset), Some(lastOffset)) + } else { + Seq(Some(lastOffset)) + } + info(s"Loading producer state till offset $lastOffset with message format version $messageFormatVersion") + + // We want to avoid unnecessary scanning of the log to build the producer state when the broker is being + // upgraded. The basic idea is to use the absence of producer snapshot files to detect the upgrade case, + // but we have to be careful not to assume too much in the presence of broker failures. The two most common + // upgrade cases in which we expect to find no snapshots are the following: + // + // 1. The broker has been upgraded, but the topic is still on the old message format. + // 2. The broker has been upgraded, the topic is on the new message format, and we had a clean shutdown. + // + // If we hit either of these cases, we skip producer state loading and write a new snapshot at the log end + // offset (see below). The next time the log is reloaded, we will load producer state using this snapshot + // (or later snapshots). Otherwise, if there is no snapshot file, then we have to rebuild producer state + // from the first segment. + if (messageFormatVersion < RecordBatch.MAGIC_VALUE_V2 || + (producerStateManager.latestSnapshotOffset.isEmpty && reloadFromCleanShutdown)) { + // To avoid an expensive scan through all of the segments, we take empty snapshots from the start of the + // last two segments and the last offset. This should avoid the full scan in the case that the log needs + // truncation. + offsetsToSnapshot.flatten.foreach { offset => + producerStateManager.updateMapEndOffset(offset) + producerStateManager.takeSnapshot() + } + } else { + val isEmptyBeforeTruncation = producerStateManager.isEmpty && producerStateManager.mapEndOffset >= lastOffset + producerStateManager.truncateAndReload(logStartOffset, lastOffset, time.milliseconds()) + + // Only do the potentially expensive reloading if the last snapshot offset is lower than the log end + // offset (which would be the case on first startup) and there were active producers prior to truncation + // (which could be the case if truncating after initial loading). If there weren't, then truncating + // shouldn't change that fact (although it could cause a producerId to expire earlier than expected), + // and we can skip the loading. This is an optimization for users which are not yet using + // idempotent/transactional features yet. + if (lastOffset > producerStateManager.mapEndOffset && !isEmptyBeforeTruncation) { + val segmentOfLastOffset = floorLogSegment(lastOffset) + + logSegments(producerStateManager.mapEndOffset, lastOffset).foreach { segment => + val startOffset = Utils.max(segment.baseOffset, producerStateManager.mapEndOffset, logStartOffset) + producerStateManager.updateMapEndOffset(startOffset) + + if (offsetsToSnapshot.contains(Some(segment.baseOffset))) + producerStateManager.takeSnapshot() + + val maxPosition = if (segmentOfLastOffset.contains(segment)) { + Option(segment.translateOffset(lastOffset)) + .map(_.position) + .getOrElse(segment.size) + } else { + segment.size + } + + val fetchDataInfo = segment.read(startOffset, + maxSize = Int.MaxValue, + maxPosition = maxPosition, + minOneMessage = false) + if (fetchDataInfo != null) + loadProducersFromLog(producerStateManager, fetchDataInfo.records) + } + } + producerStateManager.updateMapEndOffset(lastOffset) + producerStateManager.takeSnapshot() + } } private def loadProducerState(lastOffset: Long, reloadFromCleanShutdown: Boolean): Unit = lock synchronized { @@ -2573,84 +2645,6 @@ object Log { private def isLogFile(file: File): Boolean = file.getPath.endsWith(LogFileSuffix) - // Rebuild producer state until lastOffset. This method may be called from the recovery code path, and thus must be - // free of all side-effects, i.e. it must not update any log-specific state. - private def rebuildProducerState(log: Log, - lastOffset: Long, - reloadFromCleanShutdown: Boolean, - producerStateManager: ProducerStateManager): Unit = { - val messageFormatVersion = log.config.messageFormatVersion.recordVersion.value - val segments = log.logSegments - val offsetsToSnapshot = - if (segments.nonEmpty) { - val nextLatestSegmentBaseOffset = log.lowerSegment(segments.last.baseOffset).map(_.baseOffset) - Seq(nextLatestSegmentBaseOffset, Some(segments.last.baseOffset), Some(lastOffset)) - } else { - Seq(Some(lastOffset)) - } - - // We want to avoid unnecessary scanning of the log to build the producer state when the broker is being - // upgraded. The basic idea is to use the absence of producer snapshot files to detect the upgrade case, - // but we have to be careful not to assume too much in the presence of broker failures. The two most common - // upgrade cases in which we expect to find no snapshots are the following: - // - // 1. The broker has been upgraded, but the topic is still on the old message format. - // 2. The broker has been upgraded, the topic is on the new message format, and we had a clean shutdown. - // - // If we hit either of these cases, we skip producer state loading and write a new snapshot at the log end - // offset (see below). The next time the log is reloaded, we will load producer state using this snapshot - // (or later snapshots). Otherwise, if there is no snapshot file, then we have to rebuild producer state - // from the first segment. - if (messageFormatVersion < RecordBatch.MAGIC_VALUE_V2 || - (producerStateManager.latestSnapshotOffset.isEmpty && reloadFromCleanShutdown)) { - // To avoid an expensive scan through all of the segments, we take empty snapshots from the start of the - // last two segments and the last offset. This should avoid the full scan in the case that the log needs - // truncation. - offsetsToSnapshot.flatten.foreach { offset => - producerStateManager.updateMapEndOffset(offset) - producerStateManager.takeSnapshot() - } - } else { - val isEmptyBeforeTruncation = producerStateManager.isEmpty && producerStateManager.mapEndOffset >= lastOffset - producerStateManager.truncateAndReload(log.logStartOffset, lastOffset, log.time.milliseconds()) - - // Only do the potentially expensive reloading if the last snapshot offset is lower than the log end - // offset (which would be the case on first startup) and there were active producers prior to truncation - // (which could be the case if truncating after initial loading). If there weren't, then truncating - // shouldn't change that fact (although it could cause a producerId to expire earlier than expected), - // and we can skip the loading. This is an optimization for users which are not yet using - // idempotent/transactional features yet. - if (lastOffset > producerStateManager.mapEndOffset && !isEmptyBeforeTruncation) { - val segmentOfLastOffset = log.floorLogSegment(lastOffset) - - log.logSegments(producerStateManager.mapEndOffset, lastOffset).foreach { segment => - val startOffset = Utils.max(segment.baseOffset, producerStateManager.mapEndOffset, log.logStartOffset) - producerStateManager.updateMapEndOffset(startOffset) - - if (offsetsToSnapshot.contains(Some(segment.baseOffset))) - producerStateManager.takeSnapshot() - - val maxPosition = if (segmentOfLastOffset.contains(segment)) { - Option(segment.translateOffset(lastOffset)) - .map(_.position) - .getOrElse(segment.size) - } else { - segment.size - } - - val fetchDataInfo = segment.read(startOffset, - maxSize = Int.MaxValue, - maxPosition = maxPosition, - minOneMessage = false) - if (fetchDataInfo != null) - loadProducersFromLog(producerStateManager, fetchDataInfo.records) - } - } - producerStateManager.updateMapEndOffset(lastOffset) - producerStateManager.takeSnapshot() - } - } - private def loadProducersFromLog(producerStateManager: ProducerStateManager, records: Records): Unit = { val loadedProducers = mutable.Map.empty[Long, ProducerAppendInfo] val completedTxns = ListBuffer.empty[CompletedTxn] From ec00e9f2226b976e618b402fd54b11b8405b5502 Mon Sep 17 00:00:00 2001 From: Kowshik Prakasam Date: Fri, 11 Dec 2020 13:03:47 -0800 Subject: [PATCH 4/4] Address comments from @junrao --- core/src/main/scala/kafka/log/Log.scala | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/core/src/main/scala/kafka/log/Log.scala b/core/src/main/scala/kafka/log/Log.scala index 38b279678dceb..2a8602116f7e6 100644 --- a/core/src/main/scala/kafka/log/Log.scala +++ b/core/src/main/scala/kafka/log/Log.scala @@ -935,7 +935,7 @@ class Log(@volatile private var _dir: File, maxPosition = maxPosition, minOneMessage = false) if (fetchDataInfo != null) - loadProducersFromLog(producerStateManager, fetchDataInfo.records) + loadProducersFromRecords(producerStateManager, fetchDataInfo.records) } } producerStateManager.updateMapEndOffset(lastOffset) @@ -2645,7 +2645,7 @@ object Log { private def isLogFile(file: File): Boolean = file.getPath.endsWith(LogFileSuffix) - private def loadProducersFromLog(producerStateManager: ProducerStateManager, records: Records): Unit = { + private def loadProducersFromRecords(producerStateManager: ProducerStateManager, records: Records): Unit = { val loadedProducers = mutable.Map.empty[Long, ProducerAppendInfo] val completedTxns = ListBuffer.empty[CompletedTxn] records.batches.forEach { batch =>