Remove topicId argument from isStrayReplica

Author: mimaison

core/src/main/scala/kafka/server/metadata/BrokerMetadataPublisher.scala

@@ -351,7 +351,7 @@ class BrokerMetadataPublisher(
             true
           } else {
             val replicas = newImage.topics.partitionReplicas(log.topicId.get, log.topicPartition.partition)
-            JLogManager.isStrayReplica(replicas, brokerId, log.topicId.get, log)
+            JLogManager.isStrayReplica(replicas, brokerId, log)
           }
         }
       )

storage/src/main/java/org/apache/kafka/storage/internals/log/LogManager.java

@@ -16,8 +16,6 @@
  */
 package org.apache.kafka.storage.internals.log;
 
-import org.apache.kafka.common.Uuid;
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -62,9 +60,9 @@ public static boolean waitForAllToComplete(List<Future<?>> jobs, Consumer<Throwa
      * @param log            The log object to check
      * @return true if the log should not exist on the broker, false otherwise.
      */
-    public static boolean isStrayReplica(List<Integer> replicas, int brokerId, Uuid topicId, UnifiedLog log) {
+    public static boolean isStrayReplica(List<Integer> replicas, int brokerId, UnifiedLog log) {
         if (replicas.isEmpty()) {
-            LOG.info("Found stray log dir {}: the topicId {} does not exist in the metadata image.", log, topicId);
+            LOG.info("Found stray log dir {}: the topicId {} does not exist in the metadata image.", log, log.topicId().get());
             return true;
         }
         if (!replicas.contains(brokerId)) {

storage/src/test/java/org/apache/kafka/storage/internals/log/LogManagerTest.java

@@ -21,6 +21,7 @@
 import org.junit.jupiter.api.Test;
 
 import java.util.List;
+import java.util.Optional;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -67,8 +68,9 @@ public void testWaitForAllToComplete() throws ExecutionException, InterruptedExc
     @Test
     public void testIsStrayReplica() {
         UnifiedLog log = mock(UnifiedLog.class);
-        assertTrue(LogManager.isStrayReplica(List.of(), 0, Uuid.ONE_UUID, log));
-        assertTrue(LogManager.isStrayReplica(List.of(1, 2, 3), 0, Uuid.ONE_UUID, log));
-        assertFalse(LogManager.isStrayReplica(List.of(0, 1, 2), 0, Uuid.ONE_UUID, log));
+        when(log.topicId()).thenReturn(Optional.of(Uuid.ONE_UUID));
+        assertTrue(LogManager.isStrayReplica(List.of(), 0, log));
+        assertTrue(LogManager.isStrayReplica(List.of(1, 2, 3), 0, log));
+        assertFalse(LogManager.isStrayReplica(List.of(0, 1, 2), 0, log));
     }
 }