Skip to content

KAFKA-10832: Fix Log to use the correct ProducerStateManager instance when updating producers #9718

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
junrao merged 4 commits into from
Dec 12, 2020
Merged

KAFKA-10832: Fix Log to use the correct ProducerStateManager instance when updating producers #9718

Changes from 2 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
f0466b2
MINOR: Use the correct ProducerStateManager when updating producers
kowshik Dec 9, 2020
6d348bc
Address comments from @ijuma
kowshik Dec 10, 2020
5d798b7
Address comments from @junrao
kowshik Dec 11, 2020
ec00e9f
Address comments from @junrao
kowshik Dec 11, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
209 changes: 109 additions & 100 deletions core/src/main/scala/kafka/log/Log.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -864,106 +864,18 @@ 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 {
rebuildProducerState(lastOffset, reloadFromCleanShutdown, producerStateManager)
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(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)
Expand Down Expand Up @@ -1349,7 +1261,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 += _)
}

Expand Down Expand Up @@ -1456,15 +1368,6 @@ class Log(@volatile private var _dir: File,
RecordConversionStats.EMPTY, sourceCodec, targetCodec, shallowMessageCount, validBytesCount, monotonic, lastOffsetOfFirstBatch)
}

private def updateProducers(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)
*
Expand Down Expand Up @@ -2670,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,

@junrao junrao Dec 10, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This seems too large as a companion method. I am also not sure that we should pass along Log to this method since it exposes all internal states in Log, which make it harder to track usage.

@kowshik kowshik Dec 10, 2020

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That is true. Do you feel the original approach in f0466b2 was better, where, we didn't move methods into a companion object? I'd be happy to revert to that.

@junrao junrao Dec 10, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We can probably keep the two other two simpler methods (without Log as the input) as the companion methods.

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 = {

@junrao junrao Dec 11, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Perhaps this could be name loadProducersFromRecords?

@kowshik kowshik Dec 11, 2020

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done

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 {
Expand Down