diff --git a/core/src/main/scala/kafka/cluster/Partition.scala b/core/src/main/scala/kafka/cluster/Partition.scala index a138725190ad9..698fb486fdcb5 100755 --- a/core/src/main/scala/kafka/cluster/Partition.scala +++ b/core/src/main/scala/kafka/cluster/Partition.scala @@ -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 5df8c116e157e..cba0e3802794b 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} @@ -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), @@ -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 @@ -462,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, @@ -634,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) @@ -643,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 @@ -653,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 @@ -668,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) } @@ -836,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, @@ -854,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)