KAFKA-19042 Move ConsumerTopicCreationTest to client-integration-tests module (apache#19283)

This patch moves `ConsumerTopicCreationTest` to the
`client-integration-tests` and rewrite it as Java.
The patch also streamlines the test flow. 
In the Scala version, there is a producer that produces messages, but
this is not the main purpose of the `ConsumerTopicCreationTest`.

Reviewers: Ken Huang <[email protected]>, Chia-Ping Tsai
<[email protected]>

Author: frankvicky

clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ConsumerTopicCreationTest.java

@@ -0,0 +1,125 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.consumer;
+
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.TopicListing;
+import org.apache.kafka.common.test.ClusterInstance;
+import org.apache.kafka.common.test.api.ClusterConfig;
+import org.apache.kafka.common.test.api.ClusterTemplate;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+
+import static org.apache.kafka.clients.consumer.ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG;
+import static org.apache.kafka.clients.consumer.ConsumerConfig.GROUP_PROTOCOL_CONFIG;
+import static org.apache.kafka.common.test.api.Type.KRAFT;
+import static org.apache.kafka.server.config.ServerLogConfigs.AUTO_CREATE_TOPICS_ENABLE_CONFIG;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class ConsumerTopicCreationTest {
+    private static final String TOPIC = "topic";
+    private static final long POLL_TIMEOUT = 1000;
+
+    @ClusterTemplate("autoCreateTopicsConfigs")
+    void testAsyncConsumerTopicCreationIfConsumerAllowToCreateTopic(ClusterInstance cluster) throws Exception {
+        try (Consumer<byte[], byte[]> consumer = createConsumer(cluster, GroupProtocol.CONSUMER, true)) {
+            subscribeAndPoll(consumer);
+            assertTopicCreateBasedOnPermission(cluster);
+        }
+    }
+
+    @ClusterTemplate("autoCreateTopicsConfigs")
+    void testAsyncConsumerTopicCreationIfConsumerDisallowToCreateTopic(ClusterInstance cluster) throws Exception {
+        try (Consumer<byte[], byte[]> consumer = createConsumer(cluster, GroupProtocol.CONSUMER, false)) {
+            subscribeAndPoll(consumer);
+            assertTopicNotCreate(cluster);
+        }
+    }
+
+    @ClusterTemplate("autoCreateTopicsConfigs")
+    void testClassicConsumerTopicCreationIfConsumerAllowToCreateTopic(ClusterInstance cluster) throws Exception {
+        try (Consumer<byte[], byte[]> consumer = createConsumer(cluster, GroupProtocol.CLASSIC, true)) {
+            subscribeAndPoll(consumer);
+            assertTopicCreateBasedOnPermission(cluster);
+        }
+    }
+
+    @ClusterTemplate("autoCreateTopicsConfigs")
+    void testClassicConsumerTopicCreationIfConsumerDisallowToCreateTopic(ClusterInstance cluster) throws Exception {
+        try (Consumer<byte[], byte[]> consumer = createConsumer(cluster, GroupProtocol.CLASSIC, false)) {
+            subscribeAndPoll(consumer);
+            assertTopicNotCreate(cluster);
+        }
+    }
+
+    private void subscribeAndPoll(Consumer<byte[], byte[]> consumer) {
+        consumer.subscribe(List.of(TOPIC));
+        consumer.poll(Duration.ofMillis(POLL_TIMEOUT));
+    }
+
+    private void assertTopicCreateBasedOnPermission(ClusterInstance cluster) throws Exception {
+        if (allowAutoCreateTopics(cluster))
+            assertTopicCreate(cluster);
+        else
+            assertTopicNotCreate(cluster);
+    }
+
+    private boolean allowAutoCreateTopics(ClusterInstance cluster) {
+        return cluster.config().serverProperties().get(AUTO_CREATE_TOPICS_ENABLE_CONFIG).equals("true");
+    }
+
+    private void assertTopicCreate(ClusterInstance cluster) throws Exception {
+        assertTrue(getAllTopics(cluster).contains(TOPIC));
+    }
+
+    private void assertTopicNotCreate(ClusterInstance cluster) throws Exception {
+        assertFalse(getAllTopics(cluster).contains(TOPIC),
+            "Both " + AUTO_CREATE_TOPICS_ENABLE_CONFIG + " and " + ALLOW_AUTO_CREATE_TOPICS_CONFIG + " need to be true to create topic automatically");
+    }
+
+    private List<String> getAllTopics(ClusterInstance cluster) throws Exception {
+        try (Admin admin = cluster.admin()) {
+            return admin.listTopics().listings().get().stream().map(TopicListing::name).toList();
+        }
+    }
+
+    private static List<ClusterConfig> autoCreateTopicsConfigs() {
+        return List.of(
+            ClusterConfig.defaultBuilder()
+                .setTypes(Set.of(KRAFT))
+                .setServerProperties(Map.of(AUTO_CREATE_TOPICS_ENABLE_CONFIG, "true"))
+                .build(),
+            ClusterConfig.defaultBuilder()
+                .setTypes(Set.of(KRAFT))
+                .setServerProperties(Map.of(AUTO_CREATE_TOPICS_ENABLE_CONFIG, "false"))
+                .build()
+        );
+    }
+
+    private Consumer<byte[], byte[]> createConsumer(ClusterInstance cluster, GroupProtocol protocol, boolean allowConsumerAutoCreateTopics) {
+        Map<String, Object> consumerConfig = Map.of(
+            ALLOW_AUTO_CREATE_TOPICS_CONFIG, allowConsumerAutoCreateTopics,
+            GROUP_PROTOCOL_CONFIG, protocol.name().toLowerCase(Locale.ROOT)
+        );
+        return cluster.consumer(consumerConfig);
+    }
+}