[FLINK-32522][connectors/kafka] Kafka connector should depend on commons-collections instead of inheriting from flink

Author: chucheng92

flink-connector-kafka/pom.xml

@@ -72,6 +72,14 @@ under the License.
             <version>${kafka.version}</version>
         </dependency>
 
+        <!-- Others -->
+
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-collections4</artifactId>
+            <version>4.4</version>
+        </dependency>
+
         <!-- Tests -->
 
         <dependency>

flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java

@@ -59,7 +59,7 @@
 import org.apache.flink.util.ExceptionUtils;
 import org.apache.flink.util.SerializedValue;
 
-import org.apache.commons.collections.map.LinkedMap;
+import org.apache.commons.collections4.map.LinkedMap;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

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

@@ -30,7 +30,7 @@
 import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema;
 import org.apache.flink.util.DockerImageVersions;
 
-import org.apache.commons.collections.list.UnmodifiableList;
+import org.apache.commons.collections4.list.UnmodifiableList;
 import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.admin.NewTopic;
 import org.apache.kafka.clients.admin.TopicDescription;
@@ -250,7 +250,9 @@ public <T> KafkaSourceBuilder<T> getSourceBuilder(
     @SuppressWarnings("unchecked")
     public <K, V> Collection<ConsumerRecord<K, V>> getAllRecordsFromTopic(
             Properties properties, String topic) {
-        return UnmodifiableList.decorate(KafkaUtil.drainAllRecordsFromTopic(topic, properties));
+        List<ConsumerRecord<byte[], byte[]>> records =
+                KafkaUtil.drainAllRecordsFromTopic(topic, properties);
+        return UnmodifiableList.unmodifiableList((List) records);
     }
 
     @Override

flink-sql-connector-kafka/pom.xml

@@ -70,6 +70,7 @@ under the License.
                                     <include>org.apache.flink:flink-connector-base</include>
                                     <include>org.apache.flink:flink-connector-kafka</include>
                                     <include>org.apache.kafka:*</include>
+                                    <include>org.apache.commons:commons-collections4</include>
                                 </includes>
                             </artifactSet>
                             <filters>
@@ -91,6 +92,10 @@ under the License.
                                     <pattern>org.apache.kafka</pattern>
                                     <shadedPattern>org.apache.flink.kafka.shaded.org.apache.kafka</shadedPattern>
                                 </relocation>
+                                <relocation>
+                                    <pattern>org.apache.commons</pattern>
+                                    <shadedPattern>org.apache.flink.kafka.shaded.org.apache.commons</shadedPattern>
+                                </relocation>
                             </relocations>
                         </configuration>
                     </execution>

flink-sql-connector-kafka/src/main/resources/META-INF/NOTICE

@@ -7,3 +7,4 @@ The Apache Software Foundation (http://www.apache.org/).
 This project bundles the following dependencies under the Apache Software License 2.0. (http://www.apache.org/licenses/LICENSE-2.0.txt)
 
 - org.apache.kafka:kafka-clients:3.4.0
+- org.apache.commons:commons-collections4:4.4