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 all 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
57 changes: 30 additions & 27 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 @@ -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)
Expand All @@ -948,22 +948,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(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 +1333,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 +1440,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 +2645,34 @@ object Log {
private def isLogFile(file: File): Boolean =
file.getPath.endsWith(LogFileSuffix)

private def loadProducersFromRecords(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 {
Expand Down