JiayaoS commented on code in PR #22345:
URL: https://github.com/apache/kafka/pull/22345#discussion_r3291397831
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -2368,11 +2368,17 @@ class ReplicaManager(val config: KafkaConfig,
if (!localChanges.deletes.isEmpty) {
val deletes = localChanges.deletes.asScala
.map { tp =>
- val isCurrentLeader = Option(delta.image().getTopic(tp.topic()))
- .map(image => image.partitions().get(tp.partition()))
+ val topicImageOpt = Option(delta.image().getTopic(tp.topic()))
+ val isCurrentLeader = topicImageOpt
+ .flatMap(image => Option(image.partitions().get(tp.partition())))
.exists(partition => partition.leader == config.nodeId)
val deleteRemoteLog = delta.topicWasDeleted(tp.topic()) &&
isCurrentLeader
- new StopPartition(tp, true, deleteRemoteLog, false)
+ val topicId = topicImageOpt.map(_.id()).getOrElse(Uuid.ZERO_UUID)
Review Comment:
Use `ZERO_UUID` only as a defensive fallback here so local deletion can
continue when the previous image unexpectedly lacks the topic ID; remote
cleanup is skipped because it requires a valid topic ID.
##########
storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerTest.java:
##########
@@ -4204,18 +4200,6 @@ long findLogStartOffset(TopicIdPartition
topicIdPartition, UnifiedLog log) {
verifyNoMoreInteractions(remoteStorageManager);
}
- @Test
- public void testIsPartitionReady() throws InterruptedException {
Review Comment:
This test depended on `onLeadershipChange` populating
`topicIdByPartitionMap` before `isPartitionReady` could resolve a
`TopicIdPartition`. Since that cache has been removed, `isPartitionReady` now
resolves the topic ID through the metadata cache instead, so this
cache-population behavior is obsolete.
##########
storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerTest.java:
##########
@@ -1434,47 +1439,12 @@ public RemoteStorageManager
createRemoteStorageManager() {
}
}
- private void verifyInCache(TopicIdPartition... topicIdPartitions) {
- Arrays.stream(topicIdPartitions).forEach(topicIdPartition ->
- assertDoesNotThrow(() ->
remoteLogManager.fetchRemoteLogSegmentMetadata(topicIdPartition.topicPartition(),
0, 0L))
- );
- }
-
- private void verifyNotInCache(TopicIdPartition... topicIdPartitions) {
- Arrays.stream(topicIdPartitions).forEach(topicIdPartition ->
- assertThrows(KafkaException.class, () ->
-
remoteLogManager.fetchRemoteLogSegmentMetadata(topicIdPartition.topicPartition(),
0, 0L))
- );
- }
-
- @Test
- void testTopicIdCacheUpdates() throws RemoteStorageException {
Review Comment:
This test is removed because `RemoteLogManager` no longer maintains
`topicIdByPartitionMap`; topic IDs are resolved from the metadata cache for
normal operations and provided explicitly through `StopPartition` for
stop-partition handling.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
