Skip to content

KAFKA-7692: Fix ProduceStateManager SequenceNumber overflow. #5990

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
hachikuji merged 6 commits into from
Jan 25, 2019
Merged

KAFKA-7692: Fix ProduceStateManager SequenceNumber overflow. #5990

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
5fa856b
KAFKA-7692: Fix ProduceStateManager SequenceNumber overflow.
Dec 3, 2018
d16ed76
Add unittest for producerStateManager test.
Jan 11, 2019
f4bc8a2
Minor update.
Jan 14, 2019
69b44e4
Merge branch 'trunk' into 1202StateManagerSequenceId
Jan 14, 2019
a1f07ca
Use val instead of var.
Jan 14, 2019
4f0281d
Merge branch 'trunk' into 1202StateManagerSequenceId
Jan 15, 2019
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
6 changes: 3 additions & 3 deletions core/src/main/scala/kafka/log/ProducerStateManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -265,7 +265,7 @@ private[log] class ProducerAppendInfo(val producerId: Long,
None
}
} else {
append(batch.producerEpoch, batch.baseSequence, batch.lastSequence, batch.maxTimestamp, batch.lastOffset,
append(batch.producerEpoch, batch.baseSequence, batch.lastSequence, batch.maxTimestamp, batch.baseOffset, batch.lastOffset,
batch.isTransactional)
None
}
Expand All @@ -275,10 +275,11 @@ private[log] class ProducerAppendInfo(val producerId: Long,
firstSeq: Int,
lastSeq: Int,
lastTimestamp: Long,
firstOffset: Long,
lastOffset: Long,
isTransactional: Boolean): Unit = {
maybeValidateAppend(epoch, firstSeq)
updatedEntry.addBatch(epoch, lastSeq, lastOffset, lastSeq - firstSeq, lastTimestamp)
updatedEntry.addBatch(epoch, lastSeq, lastOffset, (lastOffset - firstOffset).toInt, lastTimestamp)

updatedEntry.currentTxnFirstOffset match {
case Some(_) if !isTransactional =>
Expand All @@ -287,7 +288,6 @@ private[log] class ProducerAppendInfo(val producerId: Long,

case None if isTransactional =>
// Began a new transaction
val firstOffset = lastOffset - (lastSeq - firstSeq)
updatedEntry.currentTxnFirstOffset = Some(firstOffset)
transactions += new TxnMetadata(producerId, firstOffset)

Expand Down
14 changes: 7 additions & 7 deletions core/src/test/scala/unit/kafka/log/ProducerStateManagerTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -191,7 +191,7 @@ class ProducerStateManagerTest extends JUnitSuite {
val offset = 992342L
val seq = 0
val producerAppendInfo = new ProducerAppendInfo(producerId, ProducerStateEntry.empty(producerId), ValidationType.Full)
producerAppendInfo.append(producerEpoch, seq, seq, time.milliseconds(), offset, isTransactional = true)
producerAppendInfo.append(producerEpoch, seq, seq, time.milliseconds(), offset, offset, isTransactional = true)

val logOffsetMetadata = new LogOffsetMetadata(messageOffset = offset, segmentBaseOffset = 990000L,
relativePositionInSegment = 234224)
Expand All @@ -207,7 +207,7 @@ class ProducerStateManagerTest extends JUnitSuite {
val offset = 992342L
val seq = 0
val producerAppendInfo = new ProducerAppendInfo(producerId, ProducerStateEntry.empty(producerId), ValidationType.Full)
producerAppendInfo.append(producerEpoch, seq, seq, time.milliseconds(), offset, isTransactional = true)
producerAppendInfo.append(producerEpoch, seq, seq, time.milliseconds(), offset, offset, isTransactional = true)

// use some other offset to simulate a follower append where the log offset metadata won't typically
// match any of the transaction first offsets
Expand All @@ -224,13 +224,13 @@ class ProducerStateManagerTest extends JUnitSuite {
val producerEpoch = 0.toShort

val appendInfo = stateManager.prepareUpdate(producerId, isFromClient = true)
appendInfo.append(producerEpoch, 0, 5, time.milliseconds(), 20L, isTransactional = false)
appendInfo.append(producerEpoch, 0, 5, time.milliseconds(), 15L, 20L, isTransactional = false)
assertEquals(None, stateManager.lastEntry(producerId))
stateManager.update(appendInfo)
assertTrue(stateManager.lastEntry(producerId).isDefined)

val nextAppendInfo = stateManager.prepareUpdate(producerId, isFromClient = true)
nextAppendInfo.append(producerEpoch, 6, 10, time.milliseconds(), 30L, isTransactional = false)
nextAppendInfo.append(producerEpoch, 6, 10, time.milliseconds(), 26L, 30L, isTransactional = false)
assertTrue(stateManager.lastEntry(producerId).isDefined)

var lastEntry = stateManager.lastEntry(producerId).get
Expand All @@ -253,7 +253,7 @@ class ProducerStateManagerTest extends JUnitSuite {
append(stateManager, producerId, producerEpoch, 0, offset)

val appendInfo = stateManager.prepareUpdate(producerId, isFromClient = true)
appendInfo.append(producerEpoch, 1, 5, time.milliseconds(), 20L, isTransactional = true)
appendInfo.append(producerEpoch, 1, 5, time.milliseconds(), 16L, 20L, isTransactional = true)
var lastEntry = appendInfo.toEntry
assertEquals(producerEpoch, lastEntry.producerEpoch)
assertEquals(1, lastEntry.firstSeq)
Expand All @@ -263,7 +263,7 @@ class ProducerStateManagerTest extends JUnitSuite {
assertEquals(Some(16L), lastEntry.currentTxnFirstOffset)
assertEquals(List(new TxnMetadata(producerId, 16L)), appendInfo.startedTransactions)

appendInfo.append(producerEpoch, 6, 10, time.milliseconds(), 30L, isTransactional = true)
appendInfo.append(producerEpoch, 6, 10, time.milliseconds(), 26L, 30L, isTransactional = true)
lastEntry = appendInfo.toEntry
assertEquals(producerEpoch, lastEntry.producerEpoch)
assertEquals(1, lastEntry.firstSeq)
Expand Down Expand Up @@ -819,7 +819,7 @@ class ProducerStateManagerTest extends JUnitSuite {
isTransactional: Boolean = false,
isFromClient : Boolean = true): Unit = {
val producerAppendInfo = stateManager.prepareUpdate(producerId, isFromClient)
producerAppendInfo.append(producerEpoch, seq, seq, timestamp, offset, isTransactional)
producerAppendInfo.append(producerEpoch, seq, seq, timestamp, offset, offset, isTransactional)
stateManager.update(producerAppendInfo)
stateManager.updateMapEndOffset(offset + 1)
}
Expand Down