smjn commented on code in PR #18656:
URL: https://github.com/apache/kafka/pull/18656#discussion_r1923988490
##########
core/src/test/java/kafka/test/api/ShareConsumerTest.java:
##########
@@ -101,67 +100,64 @@
@Timeout(1200)
@Tag("integration")
+@ExtendWith(ClusterTestExtensions.class)
+@ClusterTestDefaults(
+ serverProperties = {
+ @ClusterConfigProperty(key = "auto.create.topics.enable", value =
"false"),
+ @ClusterConfigProperty(key = "group.coordinator.rebalance.protocols",
value = "classic,consumer,share"),
+ @ClusterConfigProperty(key = "group.share.enable", value = "true"),
+ @ClusterConfigProperty(key = "group.share.partition.max.record.locks",
value = "10000"),
+ @ClusterConfigProperty(key = "group.share.record.lock.duration.ms",
value = "15000"),
+ @ClusterConfigProperty(key = "offsets.topic.replication.factor", value
= "1"),
+ @ClusterConfigProperty(key = "share.coordinator.state.topic.min.isr",
value = "1"),
+ @ClusterConfigProperty(key =
"share.coordinator.state.topic.num.partitions", value = "3"),
+ @ClusterConfigProperty(key =
"share.coordinator.state.topic.replication.factor", value = "1"),
+ @ClusterConfigProperty(key = "transaction.state.log.min.isr", value =
"1"),
+ @ClusterConfigProperty(key =
"transaction.state.log.replication.factor", value = "1"),
+ @ClusterConfigProperty(key = "unstable.api.versions.enable", value =
"true")
+ },
+ types = {Type.KRAFT}
+)
public class ShareConsumerTest {
- private KafkaClusterTestKit cluster;
+ private final ClusterInstance cluster;
private final TopicPartition tp = new TopicPartition("topic", 0);
private final TopicPartition tp2 = new TopicPartition("topic2", 0);
private final TopicPartition warmupTp = new TopicPartition("warmup", 0);
private List<TopicPartition> sgsTopicPartitions;
- private Admin adminClient;
-
- @BeforeEach
- public void createCluster(TestInfo testInfo) throws Exception {
- cluster = new KafkaClusterTestKit.Builder(
- new TestKitNodes.Builder()
- .setNumBrokerNodes(1)
- .setNumControllerNodes(1)
- .build())
- .setConfigProp("auto.create.topics.enable", "false")
- .setConfigProp("group.coordinator.rebalance.protocols",
"classic,consumer,share")
- .setConfigProp("group.share.enable", "true")
- .setConfigProp("group.share.partition.max.record.locks", "10000")
- .setConfigProp("group.share.record.lock.duration.ms", "15000")
- .setConfigProp("offsets.topic.replication.factor", "1")
- .setConfigProp("share.coordinator.state.topic.min.isr", "1")
- .setConfigProp("share.coordinator.state.topic.num.partitions", "3")
- .setConfigProp("share.coordinator.state.topic.replication.factor",
"1")
- .setConfigProp("transaction.state.log.min.isr", "1")
- .setConfigProp("transaction.state.log.replication.factor", "1")
- .setConfigProp("unstable.api.versions.enable", "true")
- .build();
- cluster.format();
- cluster.startup();
- cluster.waitForActiveController();
- cluster.waitForReadyBrokers();
- createTopic("topic");
- createTopic("topic2");
- adminClient = createAdminClient();
- sgsTopicPartitions = IntStream.range(0, 3)
- .mapToObj(part -> new
TopicPartition(Topic.SHARE_GROUP_STATE_TOPIC_NAME, part))
- .toList();
- warmup();
- }
-
- @AfterEach
- public void destroyCluster() throws Exception {
- adminClient.close();
- cluster.close();
- }
-
- @Test
+ public ShareConsumerTest(ClusterInstance cluster) {
+ this.cluster = cluster;
+ }
+
+ private void setup() {
Review Comment:
@mumrah Is there something I am missing?
--
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
