chia7712 commented on code in PR #15992: URL: https://github.com/apache/kafka/pull/15992#discussion_r1606051053
########## storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerMultipleSubscriptionsTest.java: ########## @@ -49,66 +51,34 @@ import static org.mockito.Mockito.verify; @SuppressWarnings("deprecation") // Added for Scala 2.12 compatibility for usages of JavaConverters Review Comment: Do we need this annotation? ########## storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerMultipleSubscriptionsTest.java: ########## @@ -49,66 +51,34 @@ import static org.mockito.Mockito.verify; @SuppressWarnings("deprecation") // Added for Scala 2.12 compatibility for usages of JavaConverters +@Tag("integration") +@ExtendWith(ClusterTestExtensions.class) +@ClusterTestDefaults(brokers = 3) public class TopicBasedRemoteLogMetadataManagerMultipleSubscriptionsTest { + private final ClusterInstance clusterInstance; private static final int SEG_SIZE = 1024 * 1024; private final Time time = new MockTime(1); - private final TopicBasedRemoteLogMetadataManagerHarness remoteLogMetadataManagerHarness = new TopicBasedRemoteLogMetadataManagerHarness(); - private TopicBasedRemoteLogMetadataManager rlmm() { - return remoteLogMetadataManagerHarness.remoteLogMetadataManager(); + TopicBasedRemoteLogMetadataManagerMultipleSubscriptionsTest(ClusterInstance clusterInstance) { + this.clusterInstance = clusterInstance; } - @BeforeEach - public void setup() { - // Start the cluster only. - remoteLogMetadataManagerHarness.setUp(new EmptyTestInfo()); - } - - @AfterEach - public void teardown() throws IOException { - remoteLogMetadataManagerHarness.close(); - } - - @Test + @ClusterTest public void testMultiplePartitionSubscriptions() throws Exception { // Create topics. String leaderTopic = "leader"; - HashMap<Object, Seq<Object>> assignedLeaderTopicReplicas = new HashMap<>(); - List<Object> leaderTopicReplicas = new ArrayList<>(); // Set broker id 0 as the first entry which is taken as the leader. - leaderTopicReplicas.add(0); - leaderTopicReplicas.add(1); - leaderTopicReplicas.add(2); - assignedLeaderTopicReplicas.put(0, JavaConverters.asScalaBuffer(leaderTopicReplicas)); - remoteLogMetadataManagerHarness.createTopicWithAssignment(leaderTopic, - JavaConverters.mapAsScalaMap(assignedLeaderTopicReplicas), - remoteLogMetadataManagerHarness.listenerName()); + createTopic(leaderTopic, Collections.singletonMap(0, Arrays.asList(0, 1, 2))); String followerTopic = "follower"; - HashMap<Object, Seq<Object>> assignedFollowerTopicReplicas = new HashMap<>(); - List<Object> followerTopicReplicas = new ArrayList<>(); // Set broker id 1 as the first entry which is taken as the leader. - followerTopicReplicas.add(1); - followerTopicReplicas.add(2); - followerTopicReplicas.add(0); - assignedFollowerTopicReplicas.put(0, JavaConverters.asScalaBuffer(followerTopicReplicas)); - remoteLogMetadataManagerHarness.createTopicWithAssignment( - followerTopic, JavaConverters.mapAsScalaMap(assignedFollowerTopicReplicas), - remoteLogMetadataManagerHarness.listenerName()); + createTopic(followerTopic, Collections.singletonMap(0, Arrays.asList(1, 2, 0))); String topicWithNoMessages = "no-messages-topic"; - HashMap<Object, Seq<Object>> assignedTopicReplicas = new HashMap<>(); - List<Object> noMessagesTopicReplicas = new ArrayList<>(); // Set broker id 1 as the first entry which is taken as the leader. - noMessagesTopicReplicas.add(1); - noMessagesTopicReplicas.add(2); - noMessagesTopicReplicas.add(0); - assignedTopicReplicas.put(0, JavaConverters.asScalaBuffer(noMessagesTopicReplicas)); - remoteLogMetadataManagerHarness.createTopicWithAssignment( - topicWithNoMessages, JavaConverters.mapAsScalaMap(assignedTopicReplicas), - remoteLogMetadataManagerHarness.listenerName()); + createTopic(topicWithNoMessages, Collections.singletonMap(0, Arrays.asList(1, 2, 0))); Review Comment: Could you please call `ClusterInstance#waitForTopic` to make sure all topics get created? https://github.com/apache/kafka/blob/trunk/core/src/test/java/kafka/test/ClusterInstance.java#L185 -- 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