From 45f3b3b4b86edc7b3562efb492cf4e5a796d9419 Mon Sep 17 00:00:00 2001 From: Bob Barrett Date: Thu, 28 Feb 2019 09:53:41 -0800 Subject: [PATCH 1/3] KAFKA-8002: Replica reassignment to new log dir may not complete if future and current replicas segment files have different base offsets This patch fixes a bug in log dir reassignment where Partition.maybeReplaceCurrentWithFutureReplica would compare the entire LogEndOffsetMetadata of each replica to determine whether the reassignment has completed. If the active segments of both replicas have different base segments (for example, if the current replica had previously been cleaned and the future replica rolled segments at different points), the reassignment will never complete. The fix is to compare only the LogEndOffsetMetadata.messageOffset for each replica. Tested with a unit test that simulates the compacted current replica case. --- .../main/scala/kafka/cluster/Partition.scala | 6 +-- .../unit/kafka/cluster/PartitionTest.scala | 48 ++++++++++++++++++- 2 files changed, 50 insertions(+), 4 deletions(-) diff --git a/core/src/main/scala/kafka/cluster/Partition.scala b/core/src/main/scala/kafka/cluster/Partition.scala index a138725190ad9..dfb470c74602e 100755 --- a/core/src/main/scala/kafka/cluster/Partition.scala +++ b/core/src/main/scala/kafka/cluster/Partition.scala @@ -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) { logManager.replaceCurrentWithFutureLog(topicPartition) replica.log = futureReplica.log futureReplica.log = None diff --git a/core/src/test/scala/unit/kafka/cluster/PartitionTest.scala b/core/src/test/scala/unit/kafka/cluster/PartitionTest.scala index 5df8c116e157e..7a6e92bf640e7 100644 --- a/core/src/test/scala/unit/kafka/cluster/PartitionTest.scala +++ b/core/src/test/scala/unit/kafka/cluster/PartitionTest.scala @@ -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} @@ -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 + // 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 From 3fcc8663e5bf8d3ecab5b6358bbe70d9869d2e84 Mon Sep 17 00:00:00 2001 From: Bob Barrett Date: Thu, 28 Feb 2019 10:33:14 -0800 Subject: [PATCH 2/3] Fix Scala 2.11 compile error --- core/src/test/scala/unit/kafka/cluster/PartitionTest.scala | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/core/src/test/scala/unit/kafka/cluster/PartitionTest.scala b/core/src/test/scala/unit/kafka/cluster/PartitionTest.scala index 7a6e92bf640e7..0dd91c15dc0f0 100644 --- a/core/src/test/scala/unit/kafka/cluster/PartitionTest.scala +++ b/core/src/test/scala/unit/kafka/cluster/PartitionTest.scala @@ -202,7 +202,7 @@ class PartitionTest { @Test // Verify that replacement works when the replicas have the same log end offset but different base offsets in the // active segment - def testMaybeReplaceCurrentWithFutureReplicaDifferentBaseOffsets: Unit = { + 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) @@ -212,12 +212,12 @@ class PartitionTest { new SimpleRecord("k1".getBytes, "v3".getBytes), new SimpleRecord("k2".getBytes, "v4".getBytes), new SimpleRecord("k2".getBytes, "v5".getBytes), - new SimpleRecord("k2".getBytes, "v6".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), + 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 From 8d5dee0be7906146ed338ca7c7a0838faf6317e7 Mon Sep 17 00:00:00 2001 From: Bob Barrett Date: Fri, 1 Mar 2019 09:21:46 -0800 Subject: [PATCH 3/3] Rename Replica.logEndOffset to logEndOffsetMetadata and make logEndOffset return the message offset --- .../main/scala/kafka/cluster/Partition.scala | 32 +++++++++---------- .../main/scala/kafka/cluster/Replica.scala | 19 ++++++----- .../server/ReplicaAlterLogDirsThread.scala | 8 ++--- .../kafka/server/ReplicaFetcherThread.scala | 12 +++---- .../scala/kafka/server/ReplicaManager.scala | 4 +-- .../api/AdminClientIntegrationTest.scala | 6 ++-- .../unit/kafka/cluster/PartitionTest.scala | 22 ++++++------- .../unit/kafka/cluster/ReplicaTest.scala | 2 +- .../ReplicaAlterLogDirsThreadTest.scala | 12 +++---- .../server/ReplicaFetcherThreadTest.scala | 20 ++++++------ .../epoch/LeaderEpochIntegrationTest.scala | 2 +- 11 files changed, 71 insertions(+), 68 deletions(-) diff --git a/core/src/main/scala/kafka/cluster/Partition.scala b/core/src/main/scala/kafka/cluster/Partition.scala index dfb470c74602e..698fb486fdcb5 100755 --- a/core/src/main/scala/kafka/cluster/Partition.scala +++ b/core/src/main/scala/kafka/cluster/Partition.scala @@ -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.messageOffset) - if (futureReplicaLEO.contains(replica.logEndOffset.messageOffset)) { + val futureReplicaLEO = futureLocalReplica.map(_.logEndOffset) + if (futureReplicaLEO.contains(replica.logEndOffset)) { // 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.messageOffset == futureReplica.logEndOffset.messageOffset) { + if (replica.logEndOffset == futureReplica.logEndOffset) { logManager.replaceCurrentWithFutureLog(topicPartition) replica.log = futureReplica.log futureReplica.log = None @@ -380,7 +380,7 @@ class Partition(val topicPartition: TopicPartition, newAssignedReplicas.foreach(id => getOrCreateReplica(id, partitionStateInfo.isNew)) val leaderReplica = localReplicaOrException - val leaderEpochStartOffset = leaderReplica.logEndOffset.messageOffset + val leaderEpochStartOffset = leaderReplica.logEndOffset info(s"$topicPartition starts at Leader Epoch ${partitionStateInfo.basePartitionState.leaderEpoch} from " + s"offset $leaderEpochStartOffset. Previous Leader Epoch was: $leaderEpoch") @@ -399,7 +399,7 @@ class Partition(val topicPartition: TopicPartition, } val isNewLeader = !leaderReplicaIdOpt.contains(localBrokerId) - val curLeaderLogEndOffset = leaderReplica.logEndOffset.messageOffset + val curLeaderLogEndOffset = leaderReplica.logEndOffset val curTimeMs = time.milliseconds // initialize lastCaughtUpTime of replicas as well as their lastFetchTimeMs and lastFetchLeaderLogEndOffset. (assignedReplicas - leaderReplica).foreach { replica => @@ -510,7 +510,7 @@ class Partition(val topicPartition: TopicPartition, val fetchOffset = logReadResult.info.fetchOffsetMetadata.messageOffset if (!inSyncReplicas.contains(replica) && assignedReplicas.map(_.brokerId).contains(replicaId) && - replica.logEndOffset.offsetDiff(leaderHW) >= 0 && + replica.logEndOffsetMetadata.offsetDiff(leaderHW) >= 0 && leaderEpochStartOffsetOpt.exists(fetchOffset >= _)) { val newInSyncReplicas = inSyncReplicas + replica info(s"Expanding ISR from ${inSyncReplicas.map(_.brokerId).mkString(",")} " + @@ -542,9 +542,9 @@ class Partition(val topicPartition: TopicPartition, val curInSyncReplicas = inSyncReplicas if (isTraceEnabled) { - def logEndOffsetString(r: Replica) = s"broker ${r.brokerId}: ${r.logEndOffset.messageOffset}" + def logEndOffsetString(r: Replica) = s"broker ${r.brokerId}: ${r.logEndOffset}" val (ackedReplicas, awaitingReplicas) = curInSyncReplicas.partition { replica => - replica.logEndOffset.messageOffset >= requiredOffset + replica.logEndOffset >= requiredOffset } trace(s"Progress awaiting ISR acks for offset $requiredOffset: acked: ${ackedReplicas.map(logEndOffsetString)}, " + s"awaiting ${awaitingReplicas.map(logEndOffsetString)}") @@ -588,7 +588,7 @@ class Partition(val topicPartition: TopicPartition, private def maybeIncrementLeaderHW(leaderReplica: Replica, curTime: Long = time.milliseconds): Boolean = { val allLogEndOffsets = assignedReplicas.filter { replica => curTime - replica.lastCaughtUpTimeMs <= replicaLagTimeMaxMs || inSyncReplicas.contains(replica) - }.map(_.logEndOffset) + }.map(_.logEndOffsetMetadata) val newHighWatermark = allLogEndOffsets.min(new LogOffsetMetadata.OffsetOrdering) val oldHighWatermark = leaderReplica.highWatermark @@ -600,7 +600,7 @@ class Partition(val topicPartition: TopicPartition, debug(s"High watermark updated to $newHighWatermark") true } else { - def logEndOffsetString(r: Replica) = s"replica ${r.brokerId}: ${r.logEndOffset}" + def logEndOffsetString(r: Replica) = s"replica ${r.brokerId}: ${r.logEndOffsetMetadata}" debug(s"Skipping update high watermark since new hw $newHighWatermark is not larger than old hw $oldHighWatermark. " + s"All current LEOs are ${assignedReplicas.map(logEndOffsetString)}") false @@ -643,9 +643,9 @@ class Partition(val topicPartition: TopicPartition, .format(inSyncReplicas.map(_.brokerId).mkString(","), newInSyncReplicas.map(_.brokerId).mkString(","), leaderReplica.highWatermark.messageOffset, - leaderReplica.logEndOffset.messageOffset, + leaderReplica.logEndOffset, outOfSyncReplicas.map { replica => - s"(brokerId: ${replica.brokerId}, endOffset: ${replica.logEndOffset.messageOffset})" + s"(brokerId: ${replica.brokerId}, endOffset: ${replica.logEndOffset})" }.mkString(" ") ) ) @@ -685,7 +685,7 @@ class Partition(val topicPartition: TopicPartition, val candidateReplicas = inSyncReplicas - leaderReplica val laggingReplicas = candidateReplicas.filter(r => - r.logEndOffset.messageOffset != leaderReplica.logEndOffset.messageOffset && (time.milliseconds - r.lastCaughtUpTimeMs) > maxLagMs) + r.logEndOffset != leaderReplica.logEndOffset && (time.milliseconds - r.lastCaughtUpTimeMs) > maxLagMs) if (laggingReplicas.nonEmpty) debug("Lagging replicas are %s".format(laggingReplicas.map(_.brokerId).mkString(","))) @@ -716,7 +716,7 @@ class Partition(val topicPartition: TopicPartition, } catch { case e: UnexpectedAppendOffsetException => val replica = if (isFuture) futureLocalReplicaOrException else localReplicaOrException - val logEndOffset = replica.logEndOffset.messageOffset + val logEndOffset = replica.logEndOffset if (logEndOffset == replica.logStartOffset && e.firstOffset < logEndOffset && e.lastOffset >= logEndOffset) { // This may happen if the log start offset on the leader (or current replica) falls in @@ -790,7 +790,7 @@ class Partition(val topicPartition: TopicPartition, */ val initialHighWatermark = localReplica.highWatermark.messageOffset val initialLogStartOffset = localReplica.logStartOffset - val initialLogEndOffset = localReplica.logEndOffset.messageOffset + val initialLogEndOffset = localReplica.logEndOffset val initialLastStableOffset = localReplica.lastStableOffset.messageOffset val maxOffsetOpt = fetchIsolation match { @@ -827,7 +827,7 @@ class Partition(val topicPartition: TopicPartition, val lastFetchableOffset = isolationLevel match { case Some(IsolationLevel.READ_COMMITTED) => localReplica.lastStableOffset.messageOffset case Some(IsolationLevel.READ_UNCOMMITTED) => localReplica.highWatermark.messageOffset - case None => localReplica.logEndOffset.messageOffset + case None => localReplica.logEndOffset } val epochLogString = if(currentLeaderEpoch.isPresent) { diff --git a/core/src/main/scala/kafka/cluster/Replica.scala b/core/src/main/scala/kafka/cluster/Replica.scala index c66eb5c1d9142..32055d7a5e30d 100644 --- a/core/src/main/scala/kafka/cluster/Replica.scala +++ b/core/src/main/scala/kafka/cluster/Replica.scala @@ -33,7 +33,7 @@ class Replica(val brokerId: Int, @volatile private[this] var highWatermarkMetadata = new LogOffsetMetadata(initialHighWatermarkValue) // the log end offset value, kept in all replicas; // for local replica it is the log's end offset, for remote replicas its value is only updated by follower fetch - @volatile private[this] var logEndOffsetMetadata = LogOffsetMetadata.UnknownOffsetMetadata + @volatile private[this] var _logEndOffsetMetadata = LogOffsetMetadata.UnknownOffsetMetadata // the log start offset value, kept in all replicas; // for local replica it is the log's start offset, for remote replicas its value is only updated by follower fetch @volatile private[this] var _logStartOffset = Log.UnknownLogStartOffset @@ -76,7 +76,7 @@ class Replica(val brokerId: Int, _lastCaughtUpTimeMs = math.max(_lastCaughtUpTimeMs, lastFetchTimeMs) logStartOffset = logReadResult.followerLogStartOffset - logEndOffset = logReadResult.info.fetchOffsetMetadata + logEndOffsetMetadata = logReadResult.info.fetchOffsetMetadata lastFetchLeaderLogEndOffset = logReadResult.leaderLogEndOffset lastFetchTimeMs = logReadResult.fetchTimeMs } @@ -87,12 +87,12 @@ class Replica(val brokerId: Int, _lastCaughtUpTimeMs = lastCaughtUpTimeMs } - private def logEndOffset_=(newLogEndOffset: LogOffsetMetadata) { + private def logEndOffsetMetadata_=(newLogEndOffset: LogOffsetMetadata) { if (isLocal) { throw new KafkaException(s"Should not set log end offset on partition $topicPartition's local replica $brokerId") } else { - logEndOffsetMetadata = newLogEndOffset - trace(s"Setting log end offset for replica $brokerId for partition $topicPartition to [$logEndOffsetMetadata]") + _logEndOffsetMetadata = newLogEndOffset + trace(s"Setting log end offset for replica $brokerId for partition $topicPartition to [${_logEndOffsetMetadata}]") } } @@ -112,11 +112,14 @@ class Replica(val brokerId: Int, } } - def logEndOffset: LogOffsetMetadata = + def logEndOffsetMetadata: LogOffsetMetadata = if (isLocal) log.get.logEndOffsetMetadata else - logEndOffsetMetadata + _logEndOffsetMetadata + + def logEndOffset: Long = + logEndOffsetMetadata.messageOffset /** * Increment the log start offset if the new offset is greater than the previous log start offset. The replica @@ -201,7 +204,7 @@ class Replica(val brokerId: Int, def offsetSnapshot: LogOffsetSnapshot = { LogOffsetSnapshot( logStartOffset = logStartOffset, - logEndOffset = logEndOffset, + logEndOffset = logEndOffsetMetadata, highWatermark = highWatermark, lastStableOffset = lastStableOffset) } diff --git a/core/src/main/scala/kafka/server/ReplicaAlterLogDirsThread.scala b/core/src/main/scala/kafka/server/ReplicaAlterLogDirsThread.scala index 8df234e65a317..cfe1ca696c373 100644 --- a/core/src/main/scala/kafka/server/ReplicaAlterLogDirsThread.scala +++ b/core/src/main/scala/kafka/server/ReplicaAlterLogDirsThread.scala @@ -58,7 +58,7 @@ class ReplicaAlterLogDirsThread(name: String, } override protected def logEndOffset(topicPartition: TopicPartition): Long = { - replicaMgr.futureLocalReplicaOrException(topicPartition).logEndOffset.messageOffset + replicaMgr.futureLocalReplicaOrException(topicPartition).logEndOffset } override protected def endOffsetForEpoch(topicPartition: TopicPartition, epoch: Int): Option[OffsetAndEpoch] = { @@ -103,12 +103,12 @@ class ReplicaAlterLogDirsThread(name: String, val partition = replicaMgr.getPartition(topicPartition).get val records = toMemoryRecords(partitionData.records) - if (fetchOffset != futureReplica.logEndOffset.messageOffset) + if (fetchOffset != futureReplica.logEndOffset) throw new IllegalStateException("Offset mismatch for the future replica %s: fetched offset = %d, log end offset = %d.".format( - topicPartition, fetchOffset, futureReplica.logEndOffset.messageOffset)) + topicPartition, fetchOffset, futureReplica.logEndOffset)) val logAppendInfo = partition.appendRecordsToFollowerOrFutureReplica(records, isFuture = true) - val futureReplicaHighWatermark = futureReplica.logEndOffset.messageOffset.min(partitionData.highWatermark) + val futureReplicaHighWatermark = futureReplica.logEndOffset.min(partitionData.highWatermark) futureReplica.highWatermark = new LogOffsetMetadata(futureReplicaHighWatermark) futureReplica.maybeIncrementLogStartOffset(partitionData.logStartOffset) diff --git a/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala b/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala index 64ae71e35f0f0..441745d6d2846 100644 --- a/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala +++ b/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala @@ -99,7 +99,7 @@ class ReplicaFetcherThread(name: String, } override protected def logEndOffset(topicPartition: TopicPartition): Long = { - replicaMgr.localReplicaOrException(topicPartition).logEndOffset.messageOffset + replicaMgr.localReplicaOrException(topicPartition).logEndOffset } override protected def endOffsetForEpoch(topicPartition: TopicPartition, epoch: Int): Option[OffsetAndEpoch] = { @@ -145,21 +145,21 @@ class ReplicaFetcherThread(name: String, maybeWarnIfOversizedRecords(records, topicPartition) - if (fetchOffset != replica.logEndOffset.messageOffset) + if (fetchOffset != replica.logEndOffset) throw new IllegalStateException("Offset mismatch for partition %s: fetched offset = %d, log end offset = %d.".format( - topicPartition, fetchOffset, replica.logEndOffset.messageOffset)) + topicPartition, fetchOffset, replica.logEndOffset)) if (isTraceEnabled) trace("Follower has replica log end offset %d for partition %s. Received %d messages and leader hw %d" - .format(replica.logEndOffset.messageOffset, topicPartition, records.sizeInBytes, partitionData.highWatermark)) + .format(replica.logEndOffset, topicPartition, records.sizeInBytes, partitionData.highWatermark)) // Append the leader's messages to the log val logAppendInfo = partition.appendRecordsToFollowerOrFutureReplica(records, isFuture = false) if (isTraceEnabled) trace("Follower has replica log end offset %d after appending %d bytes of messages for partition %s" - .format(replica.logEndOffset.messageOffset, records.sizeInBytes, topicPartition)) - val followerHighWatermark = replica.logEndOffset.messageOffset.min(partitionData.highWatermark) + .format(replica.logEndOffset, records.sizeInBytes, topicPartition)) + val followerHighWatermark = replica.logEndOffset.min(partitionData.highWatermark) val leaderLogStartOffset = partitionData.logStartOffset // for the follower replica, we do not need to keep // its segment base offset the physical position, diff --git a/core/src/main/scala/kafka/server/ReplicaManager.scala b/core/src/main/scala/kafka/server/ReplicaManager.scala index 6142c24943d51..0aaa0bcdc5c2f 100644 --- a/core/src/main/scala/kafka/server/ReplicaManager.scala +++ b/core/src/main/scala/kafka/server/ReplicaManager.scala @@ -670,7 +670,7 @@ class ReplicaManager(val config: KafkaConfig, localReplica(topicPartition) match { case Some(replica) => if (isFuture) - replica.logEndOffset.messageOffset - logEndOffset + replica.logEndOffset - logEndOffset else math.max(replica.highWatermark.messageOffset - logEndOffset, 0) case None => @@ -1405,7 +1405,7 @@ class ReplicaManager(val config: KafkaConfig, nonOfflinePartitionsIterator.filter(_.leaderReplicaIfLocal.isDefined) def getLogEndOffset(topicPartition: TopicPartition): Option[Long] = - nonOfflinePartition(topicPartition).flatMap(_.leaderReplicaIfLocal.map(_.logEndOffset.messageOffset)) + nonOfflinePartition(topicPartition).flatMap(_.leaderReplicaIfLocal.map(_.logEndOffset)) // Flushes the highwatermark value for all partitions to the highwatermark file def checkpointHighWatermarks() { diff --git a/core/src/test/scala/integration/kafka/api/AdminClientIntegrationTest.scala b/core/src/test/scala/integration/kafka/api/AdminClientIntegrationTest.scala index 96de860338884..caa5175c14bcc 100644 --- a/core/src/test/scala/integration/kafka/api/AdminClientIntegrationTest.scala +++ b/core/src/test/scala/integration/kafka/api/AdminClientIntegrationTest.scala @@ -825,7 +825,7 @@ class AdminClientIntegrationTest extends IntegrationTestHarness with Logging { }, s"Expected follower to discover new log start offset $expectedStartOffset") TestUtils.waitUntilTrue(() => { - servers(followerIndex).replicaManager.localReplica(topicPartition).get.logEndOffset.messageOffset == expectedEndOffset + servers(followerIndex).replicaManager.localReplica(topicPartition).get.logEndOffset == expectedEndOffset }, s"Expected follower to catch up to log end offset $expectedEndOffset") } @@ -871,7 +871,7 @@ class AdminClientIntegrationTest extends IntegrationTestHarness with Logging { // make sure we are in the expected state after delete records for (i <- 0 until serverCount) { assertEquals(3, servers(i).replicaManager.localReplica(topicPartition).get.logStartOffset) - assertEquals(expectedLEO, servers(i).replicaManager.localReplica(topicPartition).get.logEndOffset.messageOffset) + assertEquals(expectedLEO, servers(i).replicaManager.localReplica(topicPartition).get.logEndOffset) } // we will create another dir just for one server @@ -886,7 +886,7 @@ class AdminClientIntegrationTest extends IntegrationTestHarness with Logging { // once replica moved, its LSO and LEO should match other replicas assertEquals(3, servers(0).replicaManager.localReplica(topicPartition).get.logStartOffset) - assertEquals(expectedLEO, servers(0).replicaManager.localReplica(topicPartition).get.logEndOffset.messageOffset) + assertEquals(expectedLEO, servers(0).replicaManager.localReplica(topicPartition).get.logEndOffset) } @Test diff --git a/core/src/test/scala/unit/kafka/cluster/PartitionTest.scala b/core/src/test/scala/unit/kafka/cluster/PartitionTest.scala index 0dd91c15dc0f0..cba0e3802794b 100644 --- a/core/src/test/scala/unit/kafka/cluster/PartitionTest.scala +++ b/core/src/test/scala/unit/kafka/cluster/PartitionTest.scala @@ -121,7 +121,7 @@ class PartitionTest { assertEquals(4, log.logEndOffset) val partition = setupPartitionWithMocks(leaderEpoch = leaderEpoch, isLeader = true, log = log) - assertEquals(Some(4), partition.leaderReplicaIfLocal.map(_.logEndOffset.messageOffset)) + assertEquals(Some(4), partition.leaderReplicaIfLocal.map(_.logEndOffset)) val epochEndOffset = partition.lastOffsetForLeaderEpoch(currentLeaderEpoch = Optional.of[Integer](leaderEpoch), leaderEpoch = leaderEpoch, fetchOnlyFromLeader = true) @@ -149,7 +149,7 @@ class PartitionTest { assertEquals(4, log.logEndOffset) val partition = setupPartitionWithMocks(leaderEpoch = leaderEpoch, isLeader = true, log = log) - assertEquals(Some(4), partition.leaderReplicaIfLocal.map(_.logEndOffset.messageOffset)) + assertEquals(Some(4), partition.leaderReplicaIfLocal.map(_.logEndOffset)) assertEquals(None, log.latestEpoch) val epochEndOffset = partition.lastOffsetForLeaderEpoch(currentLeaderEpoch = Optional.of[Integer](leaderEpoch), @@ -508,7 +508,7 @@ class PartitionTest { LogReadResult(info = fetchInfo, highWatermark = leaderReplica.highWatermark.messageOffset, leaderLogStartOffset = leaderReplica.logStartOffset, - leaderLogEndOffset = leaderReplica.logEndOffset.messageOffset, + leaderLogEndOffset = leaderReplica.logEndOffset, followerLogStartOffset = 0, fetchTimeMs = time.milliseconds, readSize = 10240, @@ -680,7 +680,7 @@ class PartitionTest { val initialLogStartOffset = 5L partition.truncateFullyAndStartAt(initialLogStartOffset, isFuture = false) assertEquals(s"Log end offset after truncate fully and start at $initialLogStartOffset:", - initialLogStartOffset, replica.logEndOffset.messageOffset) + initialLogStartOffset, replica.logEndOffset) assertEquals(s"Log start offset after truncate fully and start at $initialLogStartOffset:", initialLogStartOffset, replica.logStartOffset) @@ -689,7 +689,7 @@ class PartitionTest { // append one record with offset = 3 partition.appendRecordsToFollowerOrFutureReplica(createRecords(List(new SimpleRecord("k1".getBytes, "v1".getBytes)), baseOffset = 3L), isFuture = false) } - assertEquals(s"Log end offset should not change after failure to append", initialLogStartOffset, replica.logEndOffset.messageOffset) + assertEquals(s"Log end offset should not change after failure to append", initialLogStartOffset, replica.logEndOffset) // verify that we can append records that contain log start offset, even when first // offset < log start offset if the log is empty @@ -699,12 +699,12 @@ class PartitionTest { new SimpleRecord("k3".getBytes, "v3".getBytes)), baseOffset = newLogStartOffset) partition.appendRecordsToFollowerOrFutureReplica(records, isFuture = false) - assertEquals(s"Log end offset after append of 3 records with base offset $newLogStartOffset:", 7L, replica.logEndOffset.messageOffset) + assertEquals(s"Log end offset after append of 3 records with base offset $newLogStartOffset:", 7L, replica.logEndOffset) assertEquals(s"Log start offset after append of 3 records with base offset $newLogStartOffset:", newLogStartOffset, replica.logStartOffset) // and we can append more records after that partition.appendRecordsToFollowerOrFutureReplica(createRecords(List(new SimpleRecord("k1".getBytes, "v1".getBytes)), baseOffset = 7L), isFuture = false) - assertEquals(s"Log end offset after append of 1 record at offset 7:", 8L, replica.logEndOffset.messageOffset) + assertEquals(s"Log end offset after append of 1 record at offset 7:", 8L, replica.logEndOffset) assertEquals(s"Log start offset not expected to change:", newLogStartOffset, replica.logStartOffset) // but we cannot append to offset < log start if the log is not empty @@ -714,11 +714,11 @@ class PartitionTest { baseOffset = 3L) partition.appendRecordsToFollowerOrFutureReplica(records2, isFuture = false) } - assertEquals(s"Log end offset should not change after failure to append", 8L, replica.logEndOffset.messageOffset) + assertEquals(s"Log end offset should not change after failure to append", 8L, replica.logEndOffset) // we still can append to next offset partition.appendRecordsToFollowerOrFutureReplica(createRecords(List(new SimpleRecord("k1".getBytes, "v1".getBytes)), baseOffset = 8L), isFuture = false) - assertEquals(s"Log end offset after append of 1 record at offset 8:", 9L, replica.logEndOffset.messageOffset) + assertEquals(s"Log end offset after append of 1 record at offset 8:", 9L, replica.logEndOffset) assertEquals(s"Log start offset not expected to change:", newLogStartOffset, replica.logStartOffset) } @@ -882,7 +882,7 @@ class PartitionTest { LogReadResult(info = fetchInfo, highWatermark = leaderReplica.highWatermark.messageOffset, leaderLogStartOffset = leaderReplica.logStartOffset, - leaderLogEndOffset = leaderReplica.logEndOffset.messageOffset, + leaderLogEndOffset = leaderReplica.logEndOffset, followerLogStartOffset = 0, fetchTimeMs = time.milliseconds, readSize = 10240, @@ -900,7 +900,7 @@ class PartitionTest { assertTrue("Expected makeLeader() to return 'leader changed' after makeFollower()", partition.makeLeader(controllerEpoch, new LeaderAndIsrRequest.PartitionState( controllerEpoch, leader, leaderEpoch + 2, isr, 1, replicas, false), 2)) - val currentLeaderEpochStartOffset = leaderReplica.logEndOffset.messageOffset + val currentLeaderEpochStartOffset = leaderReplica.logEndOffset // append records with the latest leader epoch partition.appendRecordsToLeader(batch3, isFromClient = true) diff --git a/core/src/test/scala/unit/kafka/cluster/ReplicaTest.scala b/core/src/test/scala/unit/kafka/cluster/ReplicaTest.scala index b3d44682a7542..d63901a3c3462 100644 --- a/core/src/test/scala/unit/kafka/cluster/ReplicaTest.scala +++ b/core/src/test/scala/unit/kafka/cluster/ReplicaTest.scala @@ -102,7 +102,7 @@ class ReplicaTest { assertTrue(log.numberOfSegments > 5) assertEquals(0L, replica.highWatermark.messageOffset) assertEquals(0L, replica.logStartOffset) - assertEquals(100L, replica.logEndOffset.messageOffset) + assertEquals(100L, replica.logEndOffset) for (hw <- 0 to 100) { replica.highWatermark = new LogOffsetMetadata(hw) diff --git a/core/src/test/scala/unit/kafka/server/ReplicaAlterLogDirsThreadTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaAlterLogDirsThreadTest.scala index 779c0e53c54bf..9710cf22b7704 100644 --- a/core/src/test/scala/unit/kafka/server/ReplicaAlterLogDirsThreadTest.scala +++ b/core/src/test/scala/unit/kafka/server/ReplicaAlterLogDirsThreadTest.scala @@ -175,8 +175,8 @@ class ReplicaAlterLogDirsThreadTest { expect(partitionT1p0.truncateTo(capture(truncateCaptureT1p0), anyBoolean())).anyTimes() expect(partitionT1p1.truncateTo(capture(truncateCaptureT1p1), anyBoolean())).anyTimes() - expect(futureReplicaT1p0.logEndOffset).andReturn(new LogOffsetMetadata(futureReplicaLEO)).anyTimes() - expect(futureReplicaT1p1.logEndOffset).andReturn(new LogOffsetMetadata(futureReplicaLEO)).anyTimes() + expect(futureReplicaT1p0.logEndOffset).andReturn(futureReplicaLEO).anyTimes() + expect(futureReplicaT1p1.logEndOffset).andReturn(futureReplicaLEO).anyTimes() expect(futureReplicaT1p0.latestEpoch).andReturn(Some(leaderEpoch)).anyTimes() expect(futureReplicaT1p0.endOffsetForEpoch(leaderEpoch)).andReturn( @@ -246,7 +246,7 @@ class ReplicaAlterLogDirsThreadTest { expect(replicaManager.futureLocalReplicaOrException(t1p0)).andStubReturn(futureReplica) expect(partition.truncateTo(capture(truncateToCapture), EasyMock.eq(true))).anyTimes() - expect(futureReplica.logEndOffset).andReturn(new LogOffsetMetadata(futureReplicaLEO)).anyTimes() + expect(futureReplica.logEndOffset).andReturn(futureReplicaLEO).anyTimes() expect(futureReplica.latestEpoch).andReturn(Some(leaderEpoch)).once() expect(futureReplica.latestEpoch).andReturn(Some(leaderEpoch - 2)).once() @@ -315,7 +315,7 @@ class ReplicaAlterLogDirsThreadTest { expect(partition.truncateTo(capture(truncated), isFuture = EasyMock.eq(true))).anyTimes() expect(replicaManager.futureLocalReplicaOrException(t1p0)).andStubReturn(futureReplica) - expect(futureReplica.logEndOffset).andReturn(new LogOffsetMetadata(futureReplicaLEO)).anyTimes() + expect(futureReplica.logEndOffset).andReturn(futureReplicaLEO).anyTimes() expect(replicaManager.logManager).andReturn(logManager).anyTimes() // pretend this is a completely new future replica, with no leader epochs recorded @@ -372,7 +372,7 @@ class ReplicaAlterLogDirsThreadTest { expect(futureReplica.latestEpoch).andStubReturn(Some(futureReplicaLeaderEpoch)) expect(futureReplica.endOffsetForEpoch(futureReplicaLeaderEpoch)).andReturn( Some(OffsetAndEpoch(futureReplicaLEO, futureReplicaLeaderEpoch))) - expect(futureReplica.logEndOffset).andReturn(new LogOffsetMetadata(futureReplicaLEO)).anyTimes() + expect(futureReplica.logEndOffset).andReturn(futureReplicaLEO).anyTimes() expect(replicaManager.localReplica(t1p0)).andReturn(Some(replica)).anyTimes() expect(replicaManager.futureLocalReplica(t1p0)).andReturn(Some(futureReplica)).anyTimes() expect(replicaManager.futureLocalReplicaOrException(t1p0)).andReturn(futureReplica).anyTimes() @@ -453,7 +453,7 @@ class ReplicaAlterLogDirsThreadTest { expect(replicaManager.futureLocalReplicaOrException(t1p0)).andStubReturn(futureReplica) expect(futureReplica.latestEpoch).andStubReturn(Some(leaderEpoch)) - expect(futureReplica.logEndOffset).andReturn(new LogOffsetMetadata(futureReplicaLEO)).anyTimes() + expect(futureReplica.logEndOffset).andReturn(futureReplicaLEO).anyTimes() expect(futureReplica.endOffsetForEpoch(leaderEpoch)).andReturn( Some(OffsetAndEpoch(futureReplicaLEO, leaderEpoch))) expect(replicaManager.logManager).andReturn(logManager).anyTimes() diff --git a/core/src/test/scala/unit/kafka/server/ReplicaFetcherThreadTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaFetcherThreadTest.scala index 376280ab6d964..2a89a29484179 100644 --- a/core/src/test/scala/unit/kafka/server/ReplicaFetcherThreadTest.scala +++ b/core/src/test/scala/unit/kafka/server/ReplicaFetcherThreadTest.scala @@ -84,7 +84,7 @@ class ReplicaFetcherThreadTest { val leaderEpoch = 5 //Stubs - expect(replica.logEndOffset).andReturn(new LogOffsetMetadata(0)).anyTimes() + expect(replica.logEndOffset).andReturn(0).anyTimes() expect(replica.highWatermark).andReturn(new LogOffsetMetadata(0)).anyTimes() expect(replica.latestEpoch).andReturn(Some(leaderEpoch)).once() expect(replica.latestEpoch).andReturn(Some(leaderEpoch)).once() @@ -214,7 +214,7 @@ class ReplicaFetcherThreadTest { val leaderEpoch = 5 //Stubs - expect(replica.logEndOffset).andReturn(new LogOffsetMetadata(0)).anyTimes() + expect(replica.logEndOffset).andReturn(0).anyTimes() expect(replica.highWatermark).andReturn(new LogOffsetMetadata(0)).anyTimes() expect(replica.latestEpoch).andReturn(Some(leaderEpoch)).anyTimes() expect(replica.endOffsetForEpoch(leaderEpoch)).andReturn( @@ -276,7 +276,7 @@ class ReplicaFetcherThreadTest { //Stubs expect(partition.truncateTo(capture(truncateToCapture), anyBoolean())).anyTimes() - expect(replica.logEndOffset).andReturn(new LogOffsetMetadata(initialLEO)).anyTimes() + expect(replica.logEndOffset).andReturn(initialLEO).anyTimes() expect(replica.highWatermark).andReturn(new LogOffsetMetadata(initialLEO - 1)).anyTimes() expect(replica.latestEpoch).andReturn(Some(leaderEpoch)).anyTimes() expect(replica.endOffsetForEpoch(leaderEpoch)).andReturn( @@ -325,7 +325,7 @@ class ReplicaFetcherThreadTest { //Stubs expect(partition.truncateTo(capture(truncateToCapture), anyBoolean())).anyTimes() - expect(replica.logEndOffset).andReturn(new LogOffsetMetadata(initialLEO)).anyTimes() + expect(replica.logEndOffset).andReturn(initialLEO).anyTimes() expect(replica.highWatermark).andReturn(new LogOffsetMetadata(initialLEO - 3)).anyTimes() expect(replica.latestEpoch).andReturn(Some(leaderEpochAtFollower)).anyTimes() expect(replica.endOffsetForEpoch(leaderEpochAtLeader)).andReturn(None).anyTimes() @@ -375,7 +375,7 @@ class ReplicaFetcherThreadTest { // Stubs expect(partition.truncateTo(capture(truncateToCapture), anyBoolean())).anyTimes() - expect(replica.logEndOffset).andReturn(new LogOffsetMetadata(initialLEO)).anyTimes() + expect(replica.logEndOffset).andReturn(initialLEO).anyTimes() expect(replica.highWatermark).andReturn(new LogOffsetMetadata(initialLEO - 2)).anyTimes() expect(replica.latestEpoch).andReturn(Some(5)).anyTimes() expect(replica.endOffsetForEpoch(4)).andReturn( @@ -446,7 +446,7 @@ class ReplicaFetcherThreadTest { // Stubs expect(partition.truncateTo(capture(truncateToCapture), anyBoolean())).anyTimes() - expect(replica.logEndOffset).andReturn(new LogOffsetMetadata(initialLEO)).anyTimes() + expect(replica.logEndOffset).andReturn(initialLEO).anyTimes() expect(replica.highWatermark).andReturn(new LogOffsetMetadata(initialLEO - 2)).anyTimes() expect(replica.latestEpoch).andReturn(Some(5)).anyTimes() expect(replica.endOffsetForEpoch(4)).andReturn( @@ -508,7 +508,7 @@ class ReplicaFetcherThreadTest { //Stubs expect(partition.truncateTo(capture(truncated), anyBoolean())).anyTimes() - expect(replica.logEndOffset).andReturn(new LogOffsetMetadata(initialLeo)).anyTimes() + expect(replica.logEndOffset).andReturn(initialLeo).anyTimes() expect(replica.highWatermark).andReturn(new LogOffsetMetadata(initialFetchOffset)).anyTimes() expect(replica.latestEpoch).andReturn(Some(5)) expect(replicaManager.logManager).andReturn(logManager).anyTimes() @@ -553,7 +553,7 @@ class ReplicaFetcherThreadTest { //Stubs expect(replica.highWatermark).andReturn(new LogOffsetMetadata(highWaterMark)).anyTimes() expect(partition.truncateTo(capture(truncated), anyBoolean())).anyTimes() - expect(replica.logEndOffset).andReturn(new LogOffsetMetadata(initialLeo)).anyTimes() + expect(replica.logEndOffset).andReturn(initialLeo).anyTimes() expect(replica.latestEpoch).andReturn(Some(leaderEpoch)).anyTimes() // this is for the last reply with EpochEndOffset(5, 156) expect(replica.endOffsetForEpoch(leaderEpoch)).andReturn( @@ -606,7 +606,7 @@ class ReplicaFetcherThreadTest { val leaderEpoch = 4 //Stub return values - expect(replica.logEndOffset).andReturn(new LogOffsetMetadata(0)).anyTimes() + expect(replica.logEndOffset).andReturn(0).anyTimes() expect(replica.highWatermark).andReturn(new LogOffsetMetadata(0)).anyTimes() expect(replica.latestEpoch).andReturn(Some(leaderEpoch)).anyTimes() expect(replica.endOffsetForEpoch(leaderEpoch)).andReturn( @@ -657,7 +657,7 @@ class ReplicaFetcherThreadTest { //Stub return values expect(partition.truncateTo(capture(truncateToCapture), anyBoolean())).once - expect(replica.logEndOffset).andReturn(new LogOffsetMetadata(initialLEO)).anyTimes() + expect(replica.logEndOffset).andReturn(initialLEO).anyTimes() expect(replica.highWatermark).andReturn(new LogOffsetMetadata(initialLEO - 2)).anyTimes() expect(replica.latestEpoch).andReturn(Some(5)).anyTimes() expect(replica.endOffsetForEpoch(5)).andReturn(Some(OffsetAndEpoch(initialLEO, 5))).anyTimes() diff --git a/core/src/test/scala/unit/kafka/server/epoch/LeaderEpochIntegrationTest.scala b/core/src/test/scala/unit/kafka/server/epoch/LeaderEpochIntegrationTest.scala index dcb885251e9e0..cb8996cc3e882 100644 --- a/core/src/test/scala/unit/kafka/server/epoch/LeaderEpochIntegrationTest.scala +++ b/core/src/test/scala/unit/kafka/server/epoch/LeaderEpochIntegrationTest.scala @@ -145,7 +145,7 @@ class LeaderEpochIntegrationTest extends ZooKeeperTestHarness with Logging { brokers += createServer(fromProps(createBrokerConfig(101, zkConnect))) - def leo() = brokers(1).replicaManager.localReplica(tp).get.logEndOffset.messageOffset + def leo() = brokers(1).replicaManager.localReplica(tp).get.logEndOffset TestUtils.createTopic(zkClient, tp.topic, Map(tp.partition -> Seq(101)), brokers) producer = createProducer(getBrokerListStrFromServers(brokers), acks = -1)