From 553ab81d3472e0f7cae8b4dd57b4ceebfbcb9350 Mon Sep 17 00:00:00 2001 From: Jason Gustafson Date: Thu, 2 Apr 2020 11:47:14 -0700 Subject: [PATCH 1/4] KAFKA-9750; Fix race condition with log dir reassign completion --- .../kafka/server/AbstractFetcherManager.scala | 7 +- .../kafka/server/AbstractFetcherThread.scala | 7 +- .../server/ReplicaAlterLogDirsThread.scala | 21 +- .../scala/kafka/server/ReplicaManager.scala | 4 + .../server/AbstractFetcherManagerTest.scala | 2 + .../ReplicaAlterLogDirsThreadTest.scala | 239 +++++++++++++++++- 6 files changed, 259 insertions(+), 21 deletions(-) diff --git a/core/src/main/scala/kafka/server/AbstractFetcherManager.scala b/core/src/main/scala/kafka/server/AbstractFetcherManager.scala index 04cee8c7abf29..b62ed9921504d 100755 --- a/core/src/main/scala/kafka/server/AbstractFetcherManager.scala +++ b/core/src/main/scala/kafka/server/AbstractFetcherManager.scala @@ -144,10 +144,9 @@ abstract class AbstractFetcherManager[T <: AbstractFetcherThread](val name: Stri tp -> OffsetAndEpoch(brokerAndInitOffset.initOffset, brokerAndInitOffset.currentLeaderEpoch) } - fetcherThread.addPartitions(initialOffsetAndEpochs) - info(s"Added fetcher to broker ${brokerAndFetcherId.broker} for partitions $initialOffsetAndEpochs") - - failedPartitions.removeAll(partitionAndOffsets.keySet) + val addedPartitions = fetcherThread.addPartitions(initialOffsetAndEpochs) + info(s"Added fetcher to broker ${brokerAndFetcherId.broker} for partitions " + + s"${initialOffsetAndEpochs.filterKeys(addedPartitions)}") } } } diff --git a/core/src/main/scala/kafka/server/AbstractFetcherThread.scala b/core/src/main/scala/kafka/server/AbstractFetcherThread.scala index 4e4ab1e723051..bc81338ef0b57 100755 --- a/core/src/main/scala/kafka/server/AbstractFetcherThread.scala +++ b/core/src/main/scala/kafka/server/AbstractFetcherThread.scala @@ -62,7 +62,7 @@ abstract class AbstractFetcherThread(name: String, type EpochData = OffsetsForLeaderEpochRequest.PartitionData private val partitionStates = new PartitionStates[PartitionFetchState] - private val partitionMapLock = new ReentrantLock + protected val partitionMapLock = new ReentrantLock private val partitionMapCond = partitionMapLock.newCondition() private val metricId = ClientIdAndBroker(clientId, sourceBroker.host, sourceBroker.port) @@ -420,9 +420,11 @@ abstract class AbstractFetcherThread(name: String, } - def addPartitions(initialFetchStates: Map[TopicPartition, OffsetAndEpoch]): Unit = { + def addPartitions(initialFetchStates: Map[TopicPartition, OffsetAndEpoch]): Set[TopicPartition] = { partitionMapLock.lockInterruptibly() try { + failedPartitions.removeAll(initialFetchStates.keySet) + initialFetchStates.foreach { case (tp, initialFetchState) => // We can skip the truncation step iff the leader epoch matches the existing epoch val currentState = partitionStates.stateValue(tp) @@ -437,6 +439,7 @@ abstract class AbstractFetcherThread(name: String, } partitionMapCond.signalAll() + initialFetchStates.keySet } finally partitionMapLock.unlock() } diff --git a/core/src/main/scala/kafka/server/ReplicaAlterLogDirsThread.scala b/core/src/main/scala/kafka/server/ReplicaAlterLogDirsThread.scala index 43bae8f7f57db..6af78d5ac2e79 100644 --- a/core/src/main/scala/kafka/server/ReplicaAlterLogDirsThread.scala +++ b/core/src/main/scala/kafka/server/ReplicaAlterLogDirsThread.scala @@ -91,7 +91,7 @@ class ReplicaAlterLogDirsThread(name: String, Request.FutureLocalReplicaId, request.minBytes, request.maxBytes, - request.version <= 2, + false, request.fetchData.asScala.toSeq, UnboundedQuota, processResponseCallback, @@ -116,7 +116,11 @@ class ReplicaAlterLogDirsThread(name: String, throw new IllegalStateException("Offset mismatch for the future replica %s: fetched offset = %d, log end offset = %d.".format( topicPartition, fetchOffset, futureLog.logEndOffset)) - val logAppendInfo = partition.appendRecordsToFollowerOrFutureReplica(records, isFuture = true) + val logAppendInfo = if (records.sizeInBytes() > 0) + partition.appendRecordsToFollowerOrFutureReplica(records, isFuture = true) + else + None + futureLog.updateHighWatermark(partitionData.highWatermark) futureLog.maybeIncrementLogStartOffset(partitionData.logStartOffset) @@ -127,6 +131,19 @@ class ReplicaAlterLogDirsThread(name: String, logAppendInfo } + override def addPartitions(initialFetchStates: Map[TopicPartition, OffsetAndEpoch]): Set[TopicPartition] = { + partitionMapLock.lockInterruptibly() + try { + // It is possible that the log dir fetcher completed just before this call, so we + // filter only the partitions which still have a future log dir. + val filteredFetchStates = initialFetchStates.filterKeys(replicaMgr.futureLogExists) + super.addPartitions(filteredFetchStates) + filteredFetchStates.keySet + } finally { + partitionMapLock.unlock() + } + } + override protected def fetchEarliestOffsetFromLeader(topicPartition: TopicPartition, leaderEpoch: Int): Long = { val partition = replicaMgr.getPartitionOrException(topicPartition, expectLeader = false) partition.localLogOrException.logStartOffset diff --git a/core/src/main/scala/kafka/server/ReplicaManager.scala b/core/src/main/scala/kafka/server/ReplicaManager.scala index 3ae1571415753..59de8db5f46c7 100644 --- a/core/src/main/scala/kafka/server/ReplicaManager.scala +++ b/core/src/main/scala/kafka/server/ReplicaManager.scala @@ -477,6 +477,10 @@ class ReplicaManager(val config: KafkaConfig, getPartitionOrException(topicPartition, expectLeader = false).futureLocalLogOrException } + def futureLogExists(topicPartition: TopicPartition): Boolean = { + getPartitionOrException(topicPartition, expectLeader = false).futureLog.isDefined + } + def localLog(topicPartition: TopicPartition): Option[Log] = { nonOfflinePartition(topicPartition).flatMap(_.log) } diff --git a/core/src/test/scala/unit/kafka/server/AbstractFetcherManagerTest.scala b/core/src/test/scala/unit/kafka/server/AbstractFetcherManagerTest.scala index 260bbd79fce95..6f07e4401856e 100644 --- a/core/src/test/scala/unit/kafka/server/AbstractFetcherManagerTest.scala +++ b/core/src/test/scala/unit/kafka/server/AbstractFetcherManagerTest.scala @@ -58,6 +58,7 @@ class AbstractFetcherManagerTest { EasyMock.expect(fetcher.start()) EasyMock.expect(fetcher.addPartitions(Map(tp -> OffsetAndEpoch(fetchOffset, leaderEpoch)))) + .andReturn(Set(tp)) EasyMock.expect(fetcher.fetchState(tp)) .andReturn(Some(PartitionFetchState(fetchOffset, None, leaderEpoch, Truncating))) EasyMock.expect(fetcher.removePartitions(Set(tp))) @@ -116,6 +117,7 @@ class AbstractFetcherManagerTest { EasyMock.expect(fetcher.start()) EasyMock.expect(fetcher.addPartitions(Map(tp -> OffsetAndEpoch(fetchOffset, leaderEpoch)))) + .andReturn(Set(tp)) EasyMock.expect(fetcher.isThreadFailed).andReturn(true) EasyMock.replay(fetcher) diff --git a/core/src/test/scala/unit/kafka/server/ReplicaAlterLogDirsThreadTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaAlterLogDirsThreadTest.scala index 482eafd8c7a77..c2f9033ace6a9 100644 --- a/core/src/test/scala/unit/kafka/server/ReplicaAlterLogDirsThreadTest.scala +++ b/core/src/test/scala/unit/kafka/server/ReplicaAlterLogDirsThreadTest.scala @@ -18,22 +18,27 @@ package kafka.server import java.util.Optional +import kafka.api.Request import kafka.cluster.{BrokerEndPoint, Partition} import kafka.log.{Log, LogManager} import kafka.server.AbstractFetcherThread.ResultWithPartitions +import kafka.server.QuotaFactory.UnboundedQuota import kafka.utils.{DelayedItem, TestUtils} -import org.apache.kafka.common.TopicPartition import org.apache.kafka.common.errors.KafkaStorageException import org.apache.kafka.common.protocol.Errors -import org.apache.kafka.common.requests.{EpochEndOffset, OffsetsForLeaderEpochRequest} +import org.apache.kafka.common.record.MemoryRecords import org.apache.kafka.common.requests.EpochEndOffset.{UNDEFINED_EPOCH, UNDEFINED_EPOCH_OFFSET} +import org.apache.kafka.common.requests.{EpochEndOffset, FetchRequest, OffsetsForLeaderEpochRequest} +import org.apache.kafka.common.{IsolationLevel, TopicPartition} import org.easymock.EasyMock._ import org.easymock.{Capture, CaptureType, EasyMock, IExpectationSetters} import org.junit.Assert._ import org.junit.Test +import org.mockito.Mockito.{doNothing, when} +import org.mockito.{ArgumentCaptor, ArgumentMatchers, Mockito} -import scala.jdk.CollectionConverters._ import scala.collection.{Map, Seq} +import scala.jdk.CollectionConverters._ class ReplicaAlterLogDirsThreadTest { @@ -45,6 +50,207 @@ class ReplicaAlterLogDirsThreadTest { OffsetAndEpoch(offset = fetchOffset, leaderEpoch = leaderEpoch) } + @Test + def shouldNotAddPartitionIfFutureLogIsNotDefined(): Unit = { + val brokerId = 1 + val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(brokerId, "localhost:1234")) + + val replicaManager = Mockito.mock(classOf[ReplicaManager]) + val quotaManager = Mockito.mock(classOf[ReplicationQuotaManager]) + + when(replicaManager.futureLogExists(t1p0)).thenReturn(false) + + val endPoint = new BrokerEndPoint(0, "localhost", 1000) + val thread = new ReplicaAlterLogDirsThread( + "alter-logs-dirs-thread", + sourceBroker = endPoint, + brokerConfig = config, + failedPartitions = failedPartitions, + replicaMgr = replicaManager, + quota = quotaManager, + brokerTopicStats = new BrokerTopicStats) + + val addedPartitions = thread.addPartitions(Map(t1p0 -> offsetAndEpoch(0L))) + assertEquals(Set.empty, addedPartitions) + assertEquals(0, thread.partitionCount()) + assertEquals(None, thread.fetchState(t1p0)) + } + + @Test + def shouldUpdateLeaderEpochAfterFencedEpochError(): Unit = { + val brokerId = 1 + val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(brokerId, "localhost:1234")) + + val partition = Mockito.mock(classOf[Partition]) + val replicaManager = Mockito.mock(classOf[ReplicaManager]) + val quotaManager = Mockito.mock(classOf[ReplicationQuotaManager]) + val futureLog = Mockito.mock(classOf[Log]) + + val leaderEpoch = 5 + val logEndOffset = 0 + + when(replicaManager.futureLocalLogOrException(t1p0)).thenReturn(futureLog) + when(replicaManager.futureLogExists(t1p0)).thenReturn(true) + when(replicaManager.nonOfflinePartition(t1p0)).thenReturn(Some(partition)) + when(replicaManager.getPartitionOrException(t1p0, expectLeader = false)).thenReturn(partition) + + when(quotaManager.isQuotaExceeded).thenReturn(false) + + when(partition.lastOffsetForLeaderEpoch(Optional.empty(), leaderEpoch, fetchOnlyFromLeader = false)) + .thenReturn(new EpochEndOffset(leaderEpoch, logEndOffset)) + when(partition.futureLocalLogOrException).thenReturn(futureLog) + doNothing().when(partition).truncateTo(offset = 0, isFuture = true) + when(partition.maybeReplaceCurrentWithFutureReplica()).thenReturn(true) + + when(futureLog.logStartOffset).thenReturn(0L) + when(futureLog.logEndOffset).thenReturn(0L) + when(futureLog.latestEpoch).thenReturn(None) + + val fencedRequestData = new FetchRequest.PartitionData(0L, 0L, + config.replicaFetchMaxBytes, Optional.of(leaderEpoch - 1)) + val fencedResponseData = FetchPartitionData( + error = Errors.FENCED_LEADER_EPOCH, + highWatermark = -1, + logStartOffset = -1, + records = MemoryRecords.EMPTY, + lastStableOffset = None, + abortedTransactions = None, + preferredReadReplica = None, + isReassignmentFetch = false) + mockFetchFromCurrentLog(t1p0, fencedRequestData, config, replicaManager, fencedResponseData) + + val endPoint = new BrokerEndPoint(0, "localhost", 1000) + val thread = new ReplicaAlterLogDirsThread( + "alter-logs-dirs-thread", + sourceBroker = endPoint, + brokerConfig = config, + failedPartitions = failedPartitions, + replicaMgr = replicaManager, + quota = quotaManager, + brokerTopicStats = new BrokerTopicStats) + + // Initially we add the partition with an older epoch which results in an error + thread.addPartitions(Map(t1p0 -> offsetAndEpoch(fetchOffset = 0L, leaderEpoch - 1))) + assertTrue(thread.fetchState(t1p0).isDefined) + assertEquals(1, thread.partitionCount()) + + thread.doWork() + + assertTrue(failedPartitions.contains(t1p0)) + assertEquals(None, thread.fetchState(t1p0)) + assertEquals(0, thread.partitionCount()) + + // Next we update the epoch and assert that we can continue + thread.addPartitions(Map(t1p0 -> offsetAndEpoch(fetchOffset = 0L, leaderEpoch))) + assertEquals(Some(leaderEpoch), thread.fetchState(t1p0).map(_.currentLeaderEpoch)) + assertEquals(1, thread.partitionCount()) + + val requestData = new FetchRequest.PartitionData(0L, 0L, + config.replicaFetchMaxBytes, Optional.of(leaderEpoch)) + val responseData = FetchPartitionData( + error = Errors.NONE, + highWatermark = 0L, + logStartOffset = 0L, + records = MemoryRecords.EMPTY, + lastStableOffset = None, + abortedTransactions = None, + preferredReadReplica = None, + isReassignmentFetch = false) + mockFetchFromCurrentLog(t1p0, requestData, config, replicaManager, responseData) + + thread.doWork() + + assertFalse(failedPartitions.contains(t1p0)) + assertEquals(None, thread.fetchState(t1p0)) + assertEquals(0, thread.partitionCount()) + } + + @Test + def shouldReplaceCurrentLogDirWhenCaughtUp(): Unit = { + val brokerId = 1 + val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(brokerId, "localhost:1234")) + + val partition = Mockito.mock(classOf[Partition]) + val replicaManager = Mockito.mock(classOf[ReplicaManager]) + val quotaManager = Mockito.mock(classOf[ReplicationQuotaManager]) + val futureLog = Mockito.mock(classOf[Log]) + + val leaderEpoch = 5 + val logEndOffset = 0 + + when(replicaManager.futureLocalLogOrException(t1p0)).thenReturn(futureLog) + when(replicaManager.futureLogExists(t1p0)).thenReturn(true) + when(replicaManager.nonOfflinePartition(t1p0)).thenReturn(Some(partition)) + when(replicaManager.getPartitionOrException(t1p0, expectLeader = false)).thenReturn(partition) + + when(quotaManager.isQuotaExceeded).thenReturn(false) + + when(partition.lastOffsetForLeaderEpoch(Optional.empty(), leaderEpoch, fetchOnlyFromLeader = false)) + .thenReturn(new EpochEndOffset(leaderEpoch, logEndOffset)) + when(partition.futureLocalLogOrException).thenReturn(futureLog) + doNothing().when(partition).truncateTo(offset = 0, isFuture = true) + when(partition.maybeReplaceCurrentWithFutureReplica()).thenReturn(true) + + when(futureLog.logStartOffset).thenReturn(0L) + when(futureLog.logEndOffset).thenReturn(0L) + when(futureLog.latestEpoch).thenReturn(None) + + val requestData = new FetchRequest.PartitionData(0L, 0L, + config.replicaFetchMaxBytes, Optional.of(leaderEpoch)) + val responseData = FetchPartitionData( + error = Errors.NONE, + highWatermark = 0L, + logStartOffset = 0L, + records = MemoryRecords.EMPTY, + lastStableOffset = None, + abortedTransactions = None, + preferredReadReplica = None, + isReassignmentFetch = false) + mockFetchFromCurrentLog(t1p0, requestData, config, replicaManager, responseData) + + val endPoint = new BrokerEndPoint(0, "localhost", 1000) + val thread = new ReplicaAlterLogDirsThread( + "alter-logs-dirs-thread", + sourceBroker = endPoint, + brokerConfig = config, + failedPartitions = failedPartitions, + replicaMgr = replicaManager, + quota = quotaManager, + brokerTopicStats = new BrokerTopicStats) + + thread.addPartitions(Map(t1p0 -> offsetAndEpoch(fetchOffset = 0L, leaderEpoch))) + assertTrue(thread.fetchState(t1p0).isDefined) + assertEquals(1, thread.partitionCount()) + + thread.doWork() + + assertEquals(None, thread.fetchState(t1p0)) + assertEquals(0, thread.partitionCount()) + } + + private def mockFetchFromCurrentLog(topicPartition: TopicPartition, + requestData: FetchRequest.PartitionData, + config: KafkaConfig, + replicaManager: ReplicaManager, + responseData: FetchPartitionData): Unit = { + val callbackCaptor: ArgumentCaptor[Seq[(TopicPartition, FetchPartitionData)] => Unit] = + ArgumentCaptor.forClass(classOf[Seq[(TopicPartition, FetchPartitionData)] => Unit]) + when(replicaManager.fetchMessages( + timeout = ArgumentMatchers.eq(0L), + replicaId = ArgumentMatchers.eq(Request.FutureLocalReplicaId), + fetchMinBytes = ArgumentMatchers.eq(0), + fetchMaxBytes = ArgumentMatchers.eq(config.replicaFetchResponseMaxBytes), + hardMaxBytesLimit = ArgumentMatchers.eq(false), + fetchInfos = ArgumentMatchers.eq(Seq(topicPartition -> requestData)), + quota = ArgumentMatchers.eq(UnboundedQuota), + responseCallback = callbackCaptor.capture(), + isolationLevel = ArgumentMatchers.eq(IsolationLevel.READ_UNCOMMITTED), + clientMetadata = ArgumentMatchers.eq(None) + )).thenAnswer(_ => { + callbackCaptor.getValue.apply(Seq((topicPartition, responseData))) + }) + } + @Test def issuesEpochRequestFromLocalReplica(): Unit = { val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(1, "localhost:1234")) @@ -79,7 +285,7 @@ class ReplicaAlterLogDirsThreadTest { "alter-logs-dirs-thread-test1", sourceBroker = endPoint, brokerConfig = config, - failedPartitions : FailedPartitions, + failedPartitions = failedPartitions, replicaMgr = replicaManager, quota = null, brokerTopicStats = null) @@ -125,7 +331,7 @@ class ReplicaAlterLogDirsThreadTest { "alter-logs-dirs-thread-test1", sourceBroker = endPoint, brokerConfig = config, - failedPartitions: FailedPartitions, + failedPartitions = failedPartitions, replicaMgr = replicaManager, quota = null, brokerTopicStats = null) @@ -174,7 +380,9 @@ class ReplicaAlterLogDirsThreadTest { expect(replicaManager.getPartitionOrException(t1p1, expectLeader = false)) .andStubReturn(partitionT1p1) expect(replicaManager.futureLocalLogOrException(t1p0)).andStubReturn(futureLogT1p0) + expect(replicaManager.futureLogExists(t1p0)).andStubReturn(true) expect(replicaManager.futureLocalLogOrException(t1p1)).andStubReturn(futureLogT1p1) + expect(replicaManager.futureLogExists(t1p1)).andStubReturn(true) expect(partitionT1p0.truncateTo(capture(truncateCaptureT1p0), anyBoolean())).anyTimes() expect(partitionT1p1.truncateTo(capture(truncateCaptureT1p1), anyBoolean())).anyTimes() @@ -206,7 +414,7 @@ class ReplicaAlterLogDirsThreadTest { "alter-logs-dirs-thread-test1", sourceBroker = endPoint, brokerConfig = config, - failedPartitions: FailedPartitions, + failedPartitions = failedPartitions, replicaMgr = replicaManager, quota = quotaManager, brokerTopicStats = null) @@ -247,6 +455,7 @@ class ReplicaAlterLogDirsThreadTest { expect(replicaManager.getPartitionOrException(t1p0, expectLeader = false)) .andStubReturn(partition) expect(replicaManager.futureLocalLogOrException(t1p0)).andStubReturn(futureLog) + expect(replicaManager.futureLogExists(t1p0)).andStubReturn(true) expect(partition.truncateTo(capture(truncateToCapture), EasyMock.eq(true))).anyTimes() expect(futureLog.logEndOffset).andReturn(futureReplicaLEO).anyTimes() @@ -278,7 +487,7 @@ class ReplicaAlterLogDirsThreadTest { "alter-logs-dirs-thread-test1", sourceBroker = endPoint, brokerConfig = config, - failedPartitions : FailedPartitions, + failedPartitions = failedPartitions, replicaMgr = replicaManager, quota = quotaManager, brokerTopicStats = null) @@ -317,6 +526,7 @@ class ReplicaAlterLogDirsThreadTest { .andStubReturn(partition) expect(partition.truncateTo(capture(truncated), isFuture = EasyMock.eq(true))).anyTimes() expect(replicaManager.futureLocalLogOrException(t1p0)).andStubReturn(futureLog) + expect(replicaManager.futureLogExists(t1p0)).andStubReturn(true) expect(replicaManager.logManager).andReturn(logManager).anyTimes() @@ -332,7 +542,7 @@ class ReplicaAlterLogDirsThreadTest { "alter-logs-dirs-thread-test1", sourceBroker = endPoint, brokerConfig = config, - failedPartitions: FailedPartitions, + failedPartitions = failedPartitions, replicaMgr = replicaManager, quota = quotaManager, brokerTopicStats = null) @@ -372,12 +582,12 @@ class ReplicaAlterLogDirsThreadTest { expect(partition.truncateTo(capture(truncated), isFuture = EasyMock.eq(true))).once() expect(replicaManager.futureLocalLogOrException(t1p0)).andStubReturn(futureLog) + expect(replicaManager.futureLogExists(t1p0)).andStubReturn(true) expect(futureLog.logEndOffset).andReturn(futureReplicaLEO).anyTimes() expect(futureLog.latestEpoch).andStubReturn(Some(futureReplicaLeaderEpoch)) expect(futureLog.endOffsetForEpoch(futureReplicaLeaderEpoch)).andReturn( Some(OffsetAndEpoch(futureReplicaLEO, futureReplicaLeaderEpoch))) expect(replicaManager.localLog(t1p0)).andReturn(Some(log)).anyTimes() - expect(replicaManager.futureLocalLogOrException(t1p0)).andReturn(futureLog).anyTimes() // this will cause fetchEpochsFromLeader return an error with undefined offset expect(partition.lastOffsetForLeaderEpoch(Optional.of(1), futureReplicaLeaderEpoch, fetchOnlyFromLeader = false)) @@ -407,7 +617,7 @@ class ReplicaAlterLogDirsThreadTest { "alter-logs-dirs-thread-test1", sourceBroker = endPoint, brokerConfig = config, - failedPartitions: FailedPartitions, + failedPartitions = failedPartitions, replicaMgr = replicaManager, quota = quotaManager, brokerTopicStats = null) @@ -452,6 +662,7 @@ class ReplicaAlterLogDirsThreadTest { expect(partition.truncateTo(futureReplicaLEO, isFuture = true)).once() expect(replicaManager.futureLocalLogOrException(t1p0)).andStubReturn(futureLog) + expect(replicaManager.futureLogExists(t1p0)).andStubReturn(true) expect(futureLog.latestEpoch).andStubReturn(Some(leaderEpoch)) expect(futureLog.logEndOffset).andStubReturn(futureReplicaLEO) expect(futureLog.endOffsetForEpoch(leaderEpoch)).andReturn( @@ -467,7 +678,7 @@ class ReplicaAlterLogDirsThreadTest { "alter-logs-dirs-thread-test1", sourceBroker = endPoint, brokerConfig = config, - failedPartitions: FailedPartitions, + failedPartitions = failedPartitions, replicaMgr = replicaManager, quota = quotaManager, brokerTopicStats = null) @@ -507,7 +718,7 @@ class ReplicaAlterLogDirsThreadTest { "alter-logs-dirs-thread-test1", sourceBroker = endPoint, brokerConfig = config, - failedPartitions: FailedPartitions, + failedPartitions = failedPartitions, replicaMgr = replicaManager, quota = quotaManager, brokerTopicStats = null) @@ -558,7 +769,7 @@ class ReplicaAlterLogDirsThreadTest { "alter-logs-dirs-thread-test1", sourceBroker = endPoint, brokerConfig = config, - failedPartitions: FailedPartitions, + failedPartitions = failedPartitions, replicaMgr = replicaManager, quota = quotaManager, brokerTopicStats = null) @@ -607,10 +818,12 @@ class ReplicaAlterLogDirsThreadTest { expect(replicaManager.localLog(t1p0)).andReturn(Some(logT1p0)).anyTimes() expect(replicaManager.localLogOrException(t1p0)).andReturn(logT1p0).anyTimes() expect(replicaManager.futureLocalLogOrException(t1p0)).andReturn(futureLog).anyTimes() + expect(replicaManager.futureLogExists(t1p0)).andStubReturn(true) expect(replicaManager.nonOfflinePartition(t1p0)).andReturn(Some(partition)).anyTimes() expect(replicaManager.localLog(t1p1)).andReturn(Some(logT1p1)).anyTimes() expect(replicaManager.localLogOrException(t1p1)).andReturn(logT1p1).anyTimes() expect(replicaManager.futureLocalLogOrException(t1p1)).andReturn(futureLog).anyTimes() + expect(replicaManager.futureLogExists(t1p1)).andStubReturn(true) expect(replicaManager.nonOfflinePartition(t1p1)).andReturn(Some(partition)).anyTimes() } From 258044128bd5afa6bc8cadb6fb1b967b5e625b00 Mon Sep 17 00:00:00 2001 From: Jason Gustafson Date: Fri, 3 Apr 2020 00:15:38 -0700 Subject: [PATCH 2/4] Add Chia's test improvements plus a minor logging optimization --- .../kafka/server/AbstractFetcherManager.scala | 13 ++++-- .../kafka/server/AbstractFetcherThread.scala | 1 - .../server/ReplicaAlterLogDirsManager.scala | 7 +++ .../server/ReplicaAlterLogDirsThread.scala | 4 +- .../kafka/server/ReplicaManagerTest.scala | 46 +++++++++++-------- 5 files changed, 47 insertions(+), 24 deletions(-) diff --git a/core/src/main/scala/kafka/server/AbstractFetcherManager.scala b/core/src/main/scala/kafka/server/AbstractFetcherManager.scala index b62ed9921504d..38457aee3aa5f 100755 --- a/core/src/main/scala/kafka/server/AbstractFetcherManager.scala +++ b/core/src/main/scala/kafka/server/AbstractFetcherManager.scala @@ -120,7 +120,8 @@ abstract class AbstractFetcherManager[T <: AbstractFetcherThread](val name: Stri BrokerAndFetcherId(brokerAndInitialFetchOffset.leader, getFetcherId(topicPartition)) } - def addAndStartFetcherThread(brokerAndFetcherId: BrokerAndFetcherId, brokerIdAndFetcherId: BrokerIdAndFetcherId): AbstractFetcherThread = { + def addAndStartFetcherThread(brokerAndFetcherId: BrokerAndFetcherId, + brokerIdAndFetcherId: BrokerIdAndFetcherId): T = { val fetcherThread = createFetcherThread(brokerAndFetcherId.fetcherId, brokerAndFetcherId.broker) fetcherThreadMap.put(brokerIdAndFetcherId, fetcherThread) fetcherThread.start() @@ -144,13 +145,17 @@ abstract class AbstractFetcherManager[T <: AbstractFetcherThread](val name: Stri tp -> OffsetAndEpoch(brokerAndInitOffset.initOffset, brokerAndInitOffset.currentLeaderEpoch) } - val addedPartitions = fetcherThread.addPartitions(initialOffsetAndEpochs) - info(s"Added fetcher to broker ${brokerAndFetcherId.broker} for partitions " + - s"${initialOffsetAndEpochs.filterKeys(addedPartitions)}") + addPartitionsToFetcherThread(fetcherThread, initialOffsetAndEpochs) } } } + protected def addPartitionsToFetcherThread(fetcherThread: T, + initialOffsetAndEpochs: collection.Map[TopicPartition, OffsetAndEpoch]): Unit = { + fetcherThread.addPartitions(initialOffsetAndEpochs) + info(s"Added fetcher to broker ${fetcherThread.sourceBroker.id} for partitions $initialOffsetAndEpochs") + } + def removeFetcherForPartitions(partitions: Set[TopicPartition]): Unit = { lock synchronized { for (fetcher <- fetcherThreadMap.values) diff --git a/core/src/main/scala/kafka/server/AbstractFetcherThread.scala b/core/src/main/scala/kafka/server/AbstractFetcherThread.scala index bc81338ef0b57..fbc27a55e7cbf 100755 --- a/core/src/main/scala/kafka/server/AbstractFetcherThread.scala +++ b/core/src/main/scala/kafka/server/AbstractFetcherThread.scala @@ -419,7 +419,6 @@ abstract class AbstractFetcherThread(name: String, warn(s"Partition $topicPartition marked as failed") } - def addPartitions(initialFetchStates: Map[TopicPartition, OffsetAndEpoch]): Set[TopicPartition] = { partitionMapLock.lockInterruptibly() try { diff --git a/core/src/main/scala/kafka/server/ReplicaAlterLogDirsManager.scala b/core/src/main/scala/kafka/server/ReplicaAlterLogDirsManager.scala index 5b6c5b18da20f..45591389d8b28 100644 --- a/core/src/main/scala/kafka/server/ReplicaAlterLogDirsManager.scala +++ b/core/src/main/scala/kafka/server/ReplicaAlterLogDirsManager.scala @@ -18,6 +18,7 @@ package kafka.server import kafka.cluster.BrokerEndPoint +import org.apache.kafka.common.TopicPartition class ReplicaAlterLogDirsManager(brokerConfig: KafkaConfig, replicaManager: ReplicaManager, @@ -34,6 +35,12 @@ class ReplicaAlterLogDirsManager(brokerConfig: KafkaConfig, quotaManager, brokerTopicStats) } + override protected def addPartitionsToFetcherThread(fetcherThread: ReplicaAlterLogDirsThread, + initialOffsetAndEpochs: collection.Map[TopicPartition, OffsetAndEpoch]): Unit = { + val addedPartitions = fetcherThread.addPartitions(initialOffsetAndEpochs) + info(s"Added log dir fetcher for partitions ${initialOffsetAndEpochs.filterKeys(addedPartitions)}") + } + def shutdown(): Unit = { info("shutting down") closeAllFetchers() diff --git a/core/src/main/scala/kafka/server/ReplicaAlterLogDirsThread.scala b/core/src/main/scala/kafka/server/ReplicaAlterLogDirsThread.scala index 6af78d5ac2e79..09b518d9ed04b 100644 --- a/core/src/main/scala/kafka/server/ReplicaAlterLogDirsThread.scala +++ b/core/src/main/scala/kafka/server/ReplicaAlterLogDirsThread.scala @@ -136,7 +136,9 @@ class ReplicaAlterLogDirsThread(name: String, try { // It is possible that the log dir fetcher completed just before this call, so we // filter only the partitions which still have a future log dir. - val filteredFetchStates = initialFetchStates.filterKeys(replicaMgr.futureLogExists) + val filteredFetchStates = initialFetchStates.filter { case (tp, _) => + replicaMgr.futureLogExists(tp) + } super.addPartitions(filteredFetchStates) filteredFetchStates.keySet } finally { diff --git a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala index 0aca3488c999d..4a06b5a14d79b 100644 --- a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala +++ b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala @@ -211,13 +211,19 @@ class ReplicaManagerTest { @Test def testFencedErrorCausedByBecomeLeader(): Unit = { + testFencedErrorCausedByBecomeLeader(0) + testFencedErrorCausedByBecomeLeader(1) + testFencedErrorCausedByBecomeLeader(10) + } + + private[this] def testFencedErrorCausedByBecomeLeader(loopEpochChange: Int): Unit = { val replicaManager = setupReplicaManagerWithMockedPurgatories(new MockTimer) try { val brokerList = Seq[Integer](0, 1).asJava val topicPartition = new TopicPartition(topic, 0) replicaManager.createPartition(topicPartition) .createLogIfNotExists(isNew = false, isFutureReplica = false, - new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints)) + new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints)) def leaderAndIsrRequest(epoch: Int): LeaderAndIsrRequest = new LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, 0, 0, brokerEpoch, Seq(new LeaderAndIsrPartitionState() @@ -234,30 +240,34 @@ class ReplicaManagerTest { replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(0), (_, _) => ()) val partition = replicaManager.getPartitionOrException(new TopicPartition(topic, 0), expectLeader = true) - .localLogOrException - assertEquals(1, replicaManager.logManager.liveLogDirs.filterNot(_ == partition.dir.getParentFile).size) + assertEquals(1, replicaManager.logManager.liveLogDirs.filterNot(_ == partition.log.get.dir.getParentFile).size) + val previousReplicaFolder = partition.log.get.dir.getParentFile // find the live and different folder - val newReplicaFolder = replicaManager.logManager.liveLogDirs.filterNot(_ == partition.dir.getParentFile).head + val newReplicaFolder = replicaManager.logManager.liveLogDirs.filterNot(_ == partition.log.get.dir.getParentFile).head assertEquals(0, replicaManager.replicaAlterLogDirsManager.fetcherThreadMap.size) replicaManager.alterReplicaLogDirs(Map(topicPartition -> newReplicaFolder.getAbsolutePath)) + // make sure the future log is created replicaManager.futureLocalLogOrException(topicPartition) assertEquals(1, replicaManager.replicaAlterLogDirsManager.fetcherThreadMap.size) - // change the epoch from 0 to 1 in order to make fenced error - replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(1), (_, _) => ()) - TestUtils.waitUntilTrue(() => replicaManager.replicaAlterLogDirsManager.fetcherThreadMap.values.forall(_.partitionCount() == 0), - s"the partition=$topicPartition should be removed from pending state") - // the partition is added to failedPartitions if fenced error happens - // if the thread is done before ReplicaManager#becomeLeaderOrFollower updates epoch,the fenced error does - // not happen and failedPartitions is empty. - if (replicaManager.replicaAlterLogDirsManager.failedPartitions.size != 0) { + (1 to loopEpochChange).foreach(epoch => replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(epoch), (_, _) => ())) + // wait for the ReplicaAlterLogDirsThread to complete + TestUtils.waitUntilTrue(() => { replicaManager.replicaAlterLogDirsManager.shutdownIdleFetcherThreads() - assertEquals(0, replicaManager.replicaAlterLogDirsManager.fetcherThreadMap.size) - // send request again - replicaManager.alterReplicaLogDirs(Map(topicPartition -> newReplicaFolder.getAbsolutePath)) - // the future folder exists so it fails to invoke thread - assertEquals(1, replicaManager.replicaAlterLogDirsManager.fetcherThreadMap.size) - } + replicaManager.replicaAlterLogDirsManager.fetcherThreadMap.isEmpty + }, s"ReplicaAlterLogDirsThread should be gone") + + // the fenced error should be recoverable + assertEquals(0, replicaManager.replicaAlterLogDirsManager.failedPartitions.size) + // the replica change is completed after retrying + assertTrue(partition.futureLog.isEmpty) + assertEquals(newReplicaFolder.getAbsolutePath, partition.log.get.dir.getParent) + // change the replica folder again + val response = replicaManager.alterReplicaLogDirs(Map(topicPartition -> previousReplicaFolder.getAbsolutePath)) + assertNotEquals(0, response.size) + response.values.foreach(assertEquals(Errors.NONE, _)) + // should succeed to invoke ReplicaAlterLogDirsThread again + assertEquals(1, replicaManager.replicaAlterLogDirsManager.fetcherThreadMap.size) } finally replicaManager.shutdown(checkpointHW = false) } From d9df4383f99770f8025c1943986f3f815ad163e3 Mon Sep 17 00:00:00 2001 From: Jason Gustafson Date: Fri, 3 Apr 2020 00:19:45 -0700 Subject: [PATCH 3/4] Remove redunant return line --- core/src/main/scala/kafka/server/ReplicaAlterLogDirsThread.scala | 1 - 1 file changed, 1 deletion(-) diff --git a/core/src/main/scala/kafka/server/ReplicaAlterLogDirsThread.scala b/core/src/main/scala/kafka/server/ReplicaAlterLogDirsThread.scala index 09b518d9ed04b..adeddc0bb2854 100644 --- a/core/src/main/scala/kafka/server/ReplicaAlterLogDirsThread.scala +++ b/core/src/main/scala/kafka/server/ReplicaAlterLogDirsThread.scala @@ -140,7 +140,6 @@ class ReplicaAlterLogDirsThread(name: String, replicaMgr.futureLogExists(tp) } super.addPartitions(filteredFetchStates) - filteredFetchStates.keySet } finally { partitionMapLock.unlock() } From 9dd4ec7253154a2e2db31c28c0ba188434d20199 Mon Sep 17 00:00:00 2001 From: Jason Gustafson Date: Fri, 3 Apr 2020 09:21:00 -0700 Subject: [PATCH 4/4] Drop usage of `filterKeys` and improve logging --- .../scala/kafka/server/AbstractFetcherThread.scala | 2 +- .../kafka/server/ReplicaAlterLogDirsManager.scala | 11 ++++++++++- 2 files changed, 11 insertions(+), 2 deletions(-) diff --git a/core/src/main/scala/kafka/server/AbstractFetcherThread.scala b/core/src/main/scala/kafka/server/AbstractFetcherThread.scala index fbc27a55e7cbf..f80d6fa9d678a 100755 --- a/core/src/main/scala/kafka/server/AbstractFetcherThread.scala +++ b/core/src/main/scala/kafka/server/AbstractFetcherThread.scala @@ -199,7 +199,7 @@ abstract class AbstractFetcherThread(name: String, * - Send OffsetsForLeaderEpochRequest, retrieving the latest offset for each partition's * leader epoch. This is the offset the follower should truncate to ensure * accurate log replication. - * - Finally truncate the logs for partitions in the truncating phase and mark them + * - Finally truncate the logs for partitions in the truncating phase and mark the * truncation complete. Do this within a lock to ensure no leadership changes can * occur during truncation. */ diff --git a/core/src/main/scala/kafka/server/ReplicaAlterLogDirsManager.scala b/core/src/main/scala/kafka/server/ReplicaAlterLogDirsManager.scala index 45591389d8b28..85eaeaa89caf3 100644 --- a/core/src/main/scala/kafka/server/ReplicaAlterLogDirsManager.scala +++ b/core/src/main/scala/kafka/server/ReplicaAlterLogDirsManager.scala @@ -38,7 +38,16 @@ class ReplicaAlterLogDirsManager(brokerConfig: KafkaConfig, override protected def addPartitionsToFetcherThread(fetcherThread: ReplicaAlterLogDirsThread, initialOffsetAndEpochs: collection.Map[TopicPartition, OffsetAndEpoch]): Unit = { val addedPartitions = fetcherThread.addPartitions(initialOffsetAndEpochs) - info(s"Added log dir fetcher for partitions ${initialOffsetAndEpochs.filterKeys(addedPartitions)}") + val (addedInitialOffsets, notAddedInitialOffsets) = initialOffsetAndEpochs.partition { case (tp, _) => + addedPartitions.contains(tp) + } + + if (addedInitialOffsets.nonEmpty) + info(s"Added log dir fetcher for partitions with initial offsets $addedInitialOffsets") + + if (notAddedInitialOffsets.nonEmpty) + info(s"Failed to add log dir fetch for partitions ${notAddedInitialOffsets.keySet} " + + s"since the log dir reassignment has already completed") } def shutdown(): Unit = {