OmniaGM commented on code in PR #15136:
URL: https://github.com/apache/kafka/pull/15136#discussion_r1444457092
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -5592,6 +5592,63 @@ class ReplicaManagerTest {
}
}
+ @ParameterizedTest
+ @ValueSource(booleans = Array(true, false))
+ def testDeltaFollowerRecoverAbandonedFutureReplica(enableRemoteStorage:
Boolean): Unit = {
+ // Given
+ val localId = 1
+ val topicPartition = new TopicPartition("foo", 0)
+
+ val mockReplicaFetcherManager = mock(classOf[ReplicaFetcherManager])
+ val mockReplicaAlterLogDirsManager =
mock(classOf[ReplicaAlterLogDirsManager])
+ val replicaManager = setupReplicaManagerWithMockedPurgatories(
+ timer = new MockTimer(time),
+ brokerId = localId,
+ mockReplicaFetcherManager = Some(mockReplicaFetcherManager),
+ mockReplicaAlterLogDirsManager = Some(mockReplicaAlterLogDirsManager),
+ enableRemoteStorage = enableRemoteStorage,
+ shouldMockLog = true
+ )
+
+ val directoryId1 = Uuid.randomUuid()
+ val directoryId2 = Uuid.randomUuid()
+ val mockFutureLog = setupMockLog("/KAFKA-16082-test", isFuture = true)
+
+ val mockLogMgr = replicaManager.logManager
+ when(
+ mockLogMgr.getLog(topicPartition, isFuture = true)
+ ).thenReturn {
+ Some(mockFutureLog)
+ }
+ when(mockLogMgr.getLog(topicPartition)).thenReturn {
+ None
+ }
+
+ when(
+ mockLogMgr.directoryId(mockFutureLog.parentDir)
+ ).thenReturn {
+ Some(directoryId1)
Review Comment:
same as above
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -4831,15 +4831,15 @@ class ReplicaManagerTest {
try {
val foo0 = new TopicPartition("foo", 0)
val emptyDelta = new TopicsDelta(TopicsImage.EMPTY)
- val (fooPart, fooNew) = replicaManager.getOrCreatePartition(foo0,
emptyDelta, FOO_UUID).get
+ val (fooPart, fooNew) = replicaManager.getOrCreatePartition(foo0,
emptyDelta, FOO_UUID, false, None).get
Review Comment:
Any reason why we are not setting `partitionAssignedDirectoryId` to None as
a default in `replicaManager.getOrCreatePartition` signature?
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -5592,6 +5592,63 @@ class ReplicaManagerTest {
}
}
+ @ParameterizedTest
+ @ValueSource(booleans = Array(true, false))
+ def testDeltaFollowerRecoverAbandonedFutureReplica(enableRemoteStorage:
Boolean): Unit = {
+ // Given
+ val localId = 1
+ val topicPartition = new TopicPartition("foo", 0)
+
+ val mockReplicaFetcherManager = mock(classOf[ReplicaFetcherManager])
+ val mockReplicaAlterLogDirsManager =
mock(classOf[ReplicaAlterLogDirsManager])
+ val replicaManager = setupReplicaManagerWithMockedPurgatories(
+ timer = new MockTimer(time),
+ brokerId = localId,
+ mockReplicaFetcherManager = Some(mockReplicaFetcherManager),
+ mockReplicaAlterLogDirsManager = Some(mockReplicaAlterLogDirsManager),
+ enableRemoteStorage = enableRemoteStorage,
+ shouldMockLog = true
+ )
+
+ val directoryId1 = Uuid.randomUuid()
+ val directoryId2 = Uuid.randomUuid()
+ val mockFutureLog = setupMockLog("/KAFKA-16082-test", isFuture = true)
+
+ val mockLogMgr = replicaManager.logManager
+ when(
+ mockLogMgr.getLog(topicPartition, isFuture = true)
+ ).thenReturn {
Review Comment:
`{` is used to hold a block but in this case it is a simple value. So maybe
replace all `.thenReturn {Some(mockFutureLog)}` to
`.thenReturn(Some(mockFutureLog))`
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -5592,6 +5592,63 @@ class ReplicaManagerTest {
}
}
+ @ParameterizedTest
+ @ValueSource(booleans = Array(true, false))
+ def testDeltaFollowerRecoverAbandonedFutureReplica(enableRemoteStorage:
Boolean): Unit = {
+ // Given
+ val localId = 1
+ val topicPartition = new TopicPartition("foo", 0)
+
+ val mockReplicaFetcherManager = mock(classOf[ReplicaFetcherManager])
+ val mockReplicaAlterLogDirsManager =
mock(classOf[ReplicaAlterLogDirsManager])
+ val replicaManager = setupReplicaManagerWithMockedPurgatories(
+ timer = new MockTimer(time),
+ brokerId = localId,
+ mockReplicaFetcherManager = Some(mockReplicaFetcherManager),
+ mockReplicaAlterLogDirsManager = Some(mockReplicaAlterLogDirsManager),
+ enableRemoteStorage = enableRemoteStorage,
+ shouldMockLog = true
+ )
+
+ val directoryId1 = Uuid.randomUuid()
+ val directoryId2 = Uuid.randomUuid()
+ val mockFutureLog = setupMockLog("/KAFKA-16082-test", isFuture = true)
+
+ val mockLogMgr = replicaManager.logManager
+ when(
+ mockLogMgr.getLog(topicPartition, isFuture = true)
+ ).thenReturn {
+ Some(mockFutureLog)
+ }
+ when(mockLogMgr.getLog(topicPartition)).thenReturn {
+ None
Review Comment:
Same `.thenReturn {None}` to `.thenReturn(None)`
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -2773,6 +2776,12 @@ class ReplicaManager(val config: KafkaConfig,
Some(partition, false)
case HostedPartition.None =>
+ var isNew = true
Review Comment:
`isNew` can be a val if you changed the code to the following
```
val isNew = if (isLocalFollower) {
logManager.maybeRecoverAbandonedFutureLog(tp,
partitionAssignedDirectoryId, () => {
isNew = false
})
} else true
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
