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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
