Skip to content

KAFKA-8002: Replica reassignment to new log dir may not complete if f… #6346

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 3 commits into from
Mar 2, 2019
Merged

KAFKA-8002: Replica reassignment to new log dir may not complete if f… #6346

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
45f3b3b
KAFKA-8002: Replica reassignment to new log dir may not complete if f…
bob-barrett Feb 28, 2019
3fcc866
Fix Scala 2.11 compile error
bob-barrett Feb 28, 2019
8d5dee0
Rename Replica.logEndOffset to logEndOffsetMetadata and make logEndOf…
bob-barrett Mar 1, 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/cluster/Partition.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -321,14 +321,14 @@ class Partition(val topicPartition: TopicPartition,
// from its partitionStates if this method returns true
def maybeReplaceCurrentWithFutureReplica(): Boolean = {
val replica = localReplicaOrException
val futureReplicaLEO = futureLocalReplica.map(_.logEndOffset)
if (futureReplicaLEO.contains(replica.logEndOffset)) {
val futureReplicaLEO = futureLocalReplica.map(_.logEndOffset.messageOffset)
if (futureReplicaLEO.contains(replica.logEndOffset.messageOffset)) {
// The write lock is needed to make sure that while ReplicaAlterDirThread checks the LEO of the
// current replica, no other thread can update LEO of the current replica via log truncation or log append operation.
inWriteLock(leaderIsrUpdateLock) {
futureLocalReplica match {
case Some(futureReplica) =>
if (replica.logEndOffset == futureReplica.logEndOffset) {
if (replica.logEndOffset.messageOffset == futureReplica.logEndOffset.messageOffset) {

@hachikuji hachikuji Feb 28, 2019

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.

One thing I was discussing with @apovzner is whether we should consider renaming logEndOffset. The bug probably would never have occurred if we had a better name. Maybe we should have two methods:

def logEndOffset: Long
def logEndOffsetMetadata: LogOffsetMetadata

@apovzner apovzner Feb 28, 2019

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.

I like the suggestion of having logEndOffsetMetadata method. Although, need to be careful not to miss any calls that would need to be renamed to
logEndOffsetMetadata (and get to the same type of bug we found here). I guess one way to make sure this does not happen is to first change the name of logEndOffset to
logEndOffsetMetadata, make sure to compile, and only after that add
logEndOffset that returns Long.

logManager.replaceCurrentWithFutureLog(topicPartition)
replica.log = futureReplica.log
futureReplica.log = None
Expand Down
48 changes: 47 additions & 1 deletion core/src/test/scala/unit/kafka/cluster/PartitionTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ import kafka.api.{ApiVersion, Request}
import kafka.common.UnexpectedAppendOffsetException
import kafka.log.{Defaults => _, _}
import kafka.server._
import kafka.utils.{CoreUtils, MockScheduler, MockTime, TestUtils}
import kafka.utils._
import kafka.zk.KafkaZkClient
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.errors.{ApiException, OffsetNotAvailableException, ReplicaNotAvailableException}
Expand Down Expand Up @@ -199,6 +199,52 @@ class PartitionTest {
assertEquals(None, partition.futureLocalReplica)
}

@Test
// Verify that replacement works when the replicas have the same log end offset but different base offsets in the

@hachikuji hachikuji Feb 28, 2019

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.

nit: Move this comment above annotation?

// active segment
def testMaybeReplaceCurrentWithFutureReplicaDifferentBaseOffsets(): Unit = {
// Write records with duplicate keys to current replica and roll at offset 6
logManager.maybeUpdatePreferredLogDir(topicPartition, logDir1.getAbsolutePath)
val log1 = logManager.getOrCreateLog(topicPartition, logConfig)
log1.appendAsLeader(MemoryRecords.withRecords(0L, CompressionType.NONE, 0,
new SimpleRecord("k1".getBytes, "v1".getBytes),
new SimpleRecord("k1".getBytes, "v2".getBytes),
new SimpleRecord("k1".getBytes, "v3".getBytes),
new SimpleRecord("k2".getBytes, "v4".getBytes),
new SimpleRecord("k2".getBytes, "v5".getBytes),
new SimpleRecord("k2".getBytes, "v6".getBytes)
), leaderEpoch = 0)
log1.roll()
log1.appendAsLeader(MemoryRecords.withRecords(0L, CompressionType.NONE, 0,
new SimpleRecord("k3".getBytes, "v7".getBytes),
new SimpleRecord("k4".getBytes, "v8".getBytes)
), leaderEpoch = 0)

// Write to the future replica as if the log had been compacted, and do not roll the segment
logManager.maybeUpdatePreferredLogDir(topicPartition, logDir2.getAbsolutePath)
val log2 = logManager.getOrCreateLog(topicPartition, logConfig, isFuture = true)
val buffer = ByteBuffer.allocate(1024)
var builder = MemoryRecords.builder(buffer, RecordBatch.CURRENT_MAGIC_VALUE, CompressionType.NONE,
TimestampType.CREATE_TIME, 0L, RecordBatch.NO_TIMESTAMP, 0)
builder.appendWithOffset(2L, new SimpleRecord("k1".getBytes, "v3".getBytes))
builder.appendWithOffset(5L, new SimpleRecord("k2".getBytes, "v6".getBytes))
builder.appendWithOffset(6L, new SimpleRecord("k3".getBytes, "v7".getBytes))
builder.appendWithOffset(7L, new SimpleRecord("k4".getBytes, "v8".getBytes))

log2.appendAsFollower(builder.build())

val currentReplica = new Replica(brokerId, topicPartition, time, log = Some(log1))
val futureReplica = new Replica(Request.FutureLocalReplicaId, topicPartition, time, log = Some(log2))
val partition = Partition(topicPartition, time, replicaManager)

partition.addReplicaIfNotExists(futureReplica)
partition.addReplicaIfNotExists(currentReplica)
assertEquals(Some(currentReplica), partition.localReplica)
assertEquals(Some(futureReplica), partition.futureLocalReplica)

assertTrue(partition.maybeReplaceCurrentWithFutureReplica())
}

@Test
def testFetchOffsetSnapshotEpochValidationForLeader(): Unit = {
val leaderEpoch = 5
Expand Down