[FLINK-37644] Remove unused code and commons-collection

Author: AHeise

flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestBase.java

@@ -33,7 +33,6 @@
 import org.apache.flink.util.TestLogger;
 
 import com.google.common.base.MoreObjects;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.producer.Callback;
 import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.ProducerConfig;
@@ -48,17 +47,12 @@
 
 import javax.annotation.Nullable;
 
-import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Collection;
-import java.util.HashSet;
 import java.util.List;
 import java.util.Properties;
-import java.util.Set;
 import java.util.concurrent.atomic.AtomicReference;
 
-import static org.assertj.core.api.Assertions.fail;
-
 /**
  * The base for the Kafka tests. It brings up:
  *
@@ -278,96 +272,6 @@ public static <K, V> void produceToKafka(
         }
     }
 
-    /**
-     * We manually handle the timeout instead of using JUnit's timeout to return failure instead of
-     * timeout error. After timeout we assume that there are missing records and there is a bug, not
-     * that the test has run out of time.
-     */
-    public void assertAtLeastOnceForTopic(
-            Properties properties,
-            String topic,
-            int partition,
-            Set<Integer> expectedElements,
-            long timeoutMillis)
-            throws Exception {
-
-        long startMillis = System.currentTimeMillis();
-        Set<Integer> actualElements = new HashSet<>();
-
-        // until we timeout...
-        while (System.currentTimeMillis() < startMillis + timeoutMillis) {
-            properties.put(
-                    "key.deserializer",
-                    "org.apache.kafka.common.serialization.IntegerDeserializer");
-            properties.put(
-                    "value.deserializer",
-                    "org.apache.kafka.common.serialization.IntegerDeserializer");
-            // We need to set these two properties so that they are lower than request.timeout.ms.
-            // This is
-            // required for some old KafkaConsumer versions.
-            properties.put("session.timeout.ms", "2000");
-            properties.put("heartbeat.interval.ms", "500");
-
-            // query kafka for new records ...
-            Collection<ConsumerRecord<Integer, Integer>> records =
-                    kafkaServer.getAllRecordsFromTopic(properties, topic);
-
-            for (ConsumerRecord<Integer, Integer> record : records) {
-                actualElements.add(record.value());
-            }
-
-            // succeed if we got all expectedElements
-            if (actualElements.containsAll(expectedElements)) {
-                return;
-            }
-        }
-
-        fail(
-                String.format(
-                        "Expected to contain all of: <%s>, but was: <%s>",
-                        expectedElements, actualElements));
-    }
-
-    public void assertExactlyOnceForTopic(
-            Properties properties, String topic, List<Integer> expectedElements) {
-
-        List<Integer> actualElements = new ArrayList<>();
-
-        Properties consumerProperties = new Properties();
-        consumerProperties.putAll(properties);
-        consumerProperties.put(
-                "key.deserializer", "org.apache.kafka.common.serialization.IntegerDeserializer");
-        consumerProperties.put(
-                "value.deserializer", "org.apache.kafka.common.serialization.IntegerDeserializer");
-        consumerProperties.put("isolation.level", "read_committed");
-
-        // query kafka for new records ...
-        Collection<ConsumerRecord<byte[], byte[]>> records =
-                kafkaServer.getAllRecordsFromTopic(consumerProperties, topic);
-
-        for (ConsumerRecord<byte[], byte[]> record : records) {
-            actualElements.add(ByteBuffer.wrap(record.value()).getInt());
-        }
-
-        // succeed if we got all expectedElements
-        if (actualElements.equals(expectedElements)) {
-            return;
-        }
-
-        fail(
-                String.format(
-                        "Expected %s, but was: %s",
-                        formatElements(expectedElements), formatElements(actualElements)));
-    }
-
-    private String formatElements(List<Integer> elements) {
-        if (elements.size() > 50) {
-            return String.format("number of elements: <%s>", elements.size());
-        } else {
-            return String.format("elements: <%s>", elements);
-        }
-    }
-
     public static void setNumKafkaClusters(int size) {
         numKafkaClusters = size;
     }

flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironment.java

@@ -17,9 +17,6 @@
 
 package org.apache.flink.streaming.connectors.kafka;
 
-import org.apache.kafka.clients.consumer.ConsumerRecord;
-
-import java.util.Collection;
 import java.util.Map;
 import java.util.Properties;
 import java.util.UUID;
@@ -110,9 +107,6 @@ public Properties getTransactionalProducerConfig() {
         return props;
     }
 
-    public abstract <K, V> Collection<ConsumerRecord<K, V>> getAllRecordsFromTopic(
-            Properties properties, String topic);
-
     // -- offset handlers
 
     /** Simple interface to commit and retrieve offsets. */

flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java

@@ -21,12 +21,10 @@
 import org.apache.flink.connector.kafka.testutils.KafkaUtil;
 import org.apache.flink.core.testutils.CommonTestUtils;
 
-import org.apache.commons.collections.list.UnmodifiableList;
 import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.admin.NewTopic;
 import org.apache.kafka.clients.admin.TopicDescription;
 import org.apache.kafka.clients.admin.TopicListing;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.common.TopicPartition;
@@ -42,7 +40,6 @@
 
 import java.time.Duration;
 import java.util.ArrayList;
-import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -220,13 +217,6 @@ public String getVersion() {
         return DockerImageVersions.KAFKA;
     }
 
-    @Override
-    @SuppressWarnings("unchecked")
-    public <K, V> Collection<ConsumerRecord<K, V>> getAllRecordsFromTopic(
-            Properties properties, String topic) {
-        return UnmodifiableList.decorate(KafkaUtil.drainAllRecordsFromTopic(topic, properties));
-    }
-
     @Override
     public KafkaOffsetHandler createOffsetHandler() {
         return new KafkaOffsetHandlerImpl();