[FLINK-37644] Remove guava from prod code

AHeise · AHeise · commit d44f57458d41 · 2025-04-11T08:10:33.000+02:00
With Java 11 most of the usages can be easily replaced. TypeToken can be replaced by Flink's TypeExtractor. The sole hard thing was a use of Map.difference but it's only needed for logging and we can solve this in a different way.
diff --git a/flink-connector-kafka-e2e-tests/flink-end-to-end-tests-common-kafka/src/test/java/org/apache/flink/tests/util/kafka/SQLClientSchemaRegistryITCase.java b/flink-connector-kafka-e2e-tests/flink-end-to-end-tests-common-kafka/src/test/java/org/apache/flink/tests/util/kafka/SQLClientSchemaRegistryITCase.java
@@ -65,8 +65,6 @@ public class SQLClientSchemaRegistryITCase {
             ResourceTestUtils.getResource(".*avro-confluent.jar");
     private final Path sqlConnectorKafkaJar = ResourceTestUtils.getResource(".*kafka.jar");
 
-    private final Path guavaJar = ResourceTestUtils.getResource(".*guava.jar");
-
     @ClassRule public static final Network NETWORK = Network.newNetwork();
 
     @ClassRule public static final Timeout TIMEOUT = new Timeout(10, TimeUnit.MINUTES);
@@ -252,7 +250,7 @@ private List<Integer> getAllVersions(String behaviourSubject) throws Exception {
     private void executeSqlStatements(List<String> sqlLines) throws Exception {
         flink.submitSQLJob(
                 new SQLJobSubmission.SQLJobSubmissionBuilder(sqlLines)
-                        .addJars(sqlAvroJar, sqlAvroRegistryJar, sqlConnectorKafkaJar, guavaJar)
+                        .addJars(sqlAvroJar, sqlAvroRegistryJar, sqlConnectorKafkaJar)
                         .build());
     }
 }
diff --git a/flink-connector-kafka/pom.xml b/flink-connector-kafka/pom.xml
@@ -82,13 +82,14 @@ under the License.
             <version>${kafka.version}</version>
         </dependency>
 
-        <dependency>
-            <groupId>com.google.guava</groupId>
-            <artifactId>guava</artifactId>
-        </dependency>
-
         <!-- Tests -->
 
+		<dependency>
+			<groupId>com.google.guava</groupId>
+			<artifactId>guava</artifactId>
+			<scope>test</scope>
+		</dependency>
+
         <dependency>
             <groupId>org.hamcrest</groupId>
             <artifactId>hamcrest-all</artifactId>
diff --git a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/metadata/ClusterMetadata.java b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/metadata/ClusterMetadata.java
@@ -20,8 +20,6 @@
 
 import org.apache.flink.annotation.Experimental;
 
-import com.google.common.base.MoreObjects;
-
 import java.io.Serializable;
 import java.util.Objects;
 import java.util.Properties;
@@ -67,10 +65,7 @@ public Properties getProperties() {
 
     @Override
     public String toString() {
-        return MoreObjects.toStringHelper(this)
-                .add("topics", topics)
-                .add("properties", properties)
-                .toString();
+        return "ClusterMetadata{" + "topics=" + topics + ", properties=" + properties + '}';
     }
 
     @Override
diff --git a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/metadata/KafkaStream.java b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/metadata/KafkaStream.java
@@ -20,8 +20,6 @@
 
 import org.apache.flink.annotation.Experimental;
 
-import com.google.common.base.MoreObjects;
-
 import java.io.Serializable;
 import java.util.Map;
 import java.util.Objects;
@@ -68,10 +66,13 @@ public Map<String, ClusterMetadata> getClusterMetadataMap() {
 
     @Override
     public String toString() {
-        return MoreObjects.toStringHelper(this)
-                .add("streamId", streamId)
-                .add("clusterMetadataMap", clusterMetadataMap)
-                .toString();
+        return "KafkaStream{"
+                + "streamId='"
+                + streamId
+                + '\''
+                + ", clusterMetadataMap="
+                + clusterMetadataMap
+                + '}';
     }
 
     @Override
diff --git a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/MetadataUpdateEvent.java b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/MetadataUpdateEvent.java
@@ -23,8 +23,6 @@
 import org.apache.flink.connector.kafka.dynamic.metadata.KafkaStream;
 import org.apache.flink.connector.kafka.dynamic.source.reader.DynamicKafkaSourceReader;
 
-import com.google.common.base.MoreObjects;
-
 import java.util.Objects;
 import java.util.Set;
 
@@ -55,7 +53,7 @@ public Set<KafkaStream> getKafkaStreams() {
 
     @Override
     public String toString() {
-        return MoreObjects.toStringHelper(this).add("kafkaStreams", kafkaStreams).toString();
+        return "MetadataUpdateEvent{" + "kafkaStreams=" + kafkaStreams + '}';
     }
 
     @Override
diff --git a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumerator.java b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumerator.java
@@ -41,9 +41,6 @@
 import org.apache.flink.connector.kafka.source.split.KafkaPartitionSplit;
 import org.apache.flink.util.Preconditions;
 
-import com.google.common.collect.ArrayListMultimap;
-import com.google.common.collect.MapDifference;
-import com.google.common.collect.Maps;
 import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.TopicPartition;
 import org.slf4j.Logger;
@@ -61,6 +58,7 @@
 import java.util.Map.Entry;
 import java.util.Properties;
 import java.util.Set;
+import java.util.TreeMap;
 import java.util.stream.Collectors;
 
 /**
@@ -260,17 +258,11 @@ private void onHandleSubscribedStreamsFetch(Set<KafkaStream> fetchedKafkaStreams
         }
 
         if (logger.isInfoEnabled()) {
-            MapDifference<String, Set<String>> metadataDifference =
-                    Maps.difference(latestClusterTopicsMap, newClustersTopicsMap);
+            // log the maps in a sorted fashion so it's easy to see the changes
             logger.info(
-                    "Common cluster topics after metadata refresh: {}",
-                    metadataDifference.entriesInCommon());
-            logger.info(
-                    "Removed cluster topics after metadata refresh: {}",
-                    metadataDifference.entriesOnlyOnLeft());
-            logger.info(
-                    "Additional cluster topics after metadata refresh: {}",
-                    metadataDifference.entriesOnlyOnRight());
+                    "Detected changed cluster topics after metadata refresh:\nPrevious: {}\nNew: {}",
+                    new TreeMap<>(latestClusterTopicsMap),
+                    new TreeMap<>(newClustersTopicsMap));
         }
 
         DynamicKafkaSourceEnumState dynamicKafkaSourceEnumState;
@@ -456,10 +448,11 @@ public void handleSplitRequest(int subtaskId, @Nullable String requesterHostname
     public void addSplitsBack(List<DynamicKafkaSourceSplit> splits, int subtaskId) {
         logger.debug("Adding splits back for {}", subtaskId);
         // separate splits by cluster
-        ArrayListMultimap<String, KafkaPartitionSplit> kafkaPartitionSplits =
-                ArrayListMultimap.create();
+        Map<String, List<KafkaPartitionSplit>> kafkaPartitionSplits = new HashMap<>();
         for (DynamicKafkaSourceSplit split : splits) {
-            kafkaPartitionSplits.put(split.getKafkaClusterId(), split.getKafkaPartitionSplit());
+            kafkaPartitionSplits
+                    .computeIfAbsent(split.getKafkaClusterId(), unused -> new ArrayList<>())
+                    .add(split.getKafkaPartitionSplit());
         }
 
         // add splits back and assign pending splits for all enumerators
diff --git a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/subscriber/StreamPatternSubscriber.java b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/subscriber/StreamPatternSubscriber.java
@@ -22,8 +22,8 @@
 import org.apache.flink.connector.kafka.dynamic.metadata.KafkaMetadataService;
 import org.apache.flink.connector.kafka.dynamic.metadata.KafkaStream;
 
-import com.google.common.collect.ImmutableSet;
-
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Set;
 import java.util.regex.Pattern;
 
@@ -40,14 +40,14 @@ public StreamPatternSubscriber(Pattern streamPattern) {
     @Override
     public Set<KafkaStream> getSubscribedStreams(KafkaMetadataService kafkaMetadataService) {
         Set<KafkaStream> allStreams = kafkaMetadataService.getAllStreams();
-        ImmutableSet.Builder<KafkaStream> builder = ImmutableSet.builder();
+        List<KafkaStream> matches = new ArrayList<>();
         for (KafkaStream kafkaStream : allStreams) {
             String streamId = kafkaStream.getStreamId();
             if (streamPattern.matcher(streamId).find()) {
-                builder.add(kafkaStream);
+                matches.add(kafkaStream);
             }
         }
 
-        return builder.build();
+        return Set.copyOf(matches);
     }
 }
diff --git a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReader.java b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReader.java
@@ -49,7 +49,6 @@
 import org.apache.flink.util.Preconditions;
 import org.apache.flink.util.UserCodeClassLoader;
 
-import com.google.common.collect.ArrayListMultimap;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -185,10 +184,11 @@ public void addSplits(List<DynamicKafkaSourceSplit> splits) {
             return;
         }
 
-        ArrayListMultimap<String, KafkaPartitionSplit> clusterSplitsMap =
-                ArrayListMultimap.create();
+        Map<String, List<KafkaPartitionSplit>> clusterSplitsMap = new HashMap<>();
         for (DynamicKafkaSourceSplit split : splits) {
-            clusterSplitsMap.put(split.getKafkaClusterId(), split);
+            clusterSplitsMap
+                    .computeIfAbsent(split.getKafkaClusterId(), unused -> new ArrayList<>())
+                    .add(split);
         }
 
         Set<String> kafkaClusterIds = clusterSplitsMap.keySet();
diff --git a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/split/DynamicKafkaSourceSplit.java b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/split/DynamicKafkaSourceSplit.java
@@ -21,8 +21,6 @@
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.connector.kafka.source.split.KafkaPartitionSplit;
 
-import com.google.common.base.MoreObjects;
-
 import java.util.Objects;
 
 /** Split that wraps {@link KafkaPartitionSplit} with Kafka cluster information. */
@@ -56,10 +54,13 @@ public KafkaPartitionSplit getKafkaPartitionSplit() {
 
     @Override
     public String toString() {
-        return MoreObjects.toStringHelper(this)
-                .add("kafkaClusterId", kafkaClusterId)
-                .add("kafkaPartitionSplit", kafkaPartitionSplit)
-                .toString();
+        return "DynamicKafkaSourceSplit{"
+                + "kafkaClusterId='"
+                + kafkaClusterId
+                + '\''
+                + ", kafkaPartitionSplit="
+                + kafkaPartitionSplit
+                + '}';
     }
 
     @Override
diff --git a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/split/DynamicKafkaSourceSplitSerializer.java b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/split/DynamicKafkaSourceSplitSerializer.java
@@ -23,8 +23,6 @@
 import org.apache.flink.connector.kafka.source.split.KafkaPartitionSplitSerializer;
 import org.apache.flink.core.io.SimpleVersionedSerializer;
 
-import com.google.common.io.ByteStreams;
-
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.DataInputStream;
@@ -69,7 +67,7 @@ public DynamicKafkaSourceSplit deserialize(int version, byte[] serialized) throw
             int kafkaPartitionSplitSerializerVersion = in.readInt();
             KafkaPartitionSplit kafkaPartitionSplit =
                     kafkaPartitionSplitSerializer.deserialize(
-                            kafkaPartitionSplitSerializerVersion, ByteStreams.toByteArray(in));
+                            kafkaPartitionSplitSerializerVersion, in.readAllBytes());
             return new DynamicKafkaSourceSplit(kafkaClusterId, kafkaPartitionSplit);
         }
     }
diff --git a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaRecordSerializationSchemaBuilder.java b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaRecordSerializationSchemaBuilder.java
@@ -19,8 +19,8 @@
 
 import org.apache.flink.annotation.PublicEvolving;
 import org.apache.flink.api.common.serialization.SerializationSchema;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
 import org.apache.flink.connector.kafka.lineage.DefaultKafkaDatasetFacet;
 import org.apache.flink.connector.kafka.lineage.DefaultKafkaDatasetIdentifier;
 import org.apache.flink.connector.kafka.lineage.DefaultTypeDatasetFacet;
@@ -31,7 +31,6 @@
 import org.apache.flink.connector.kafka.lineage.TypeDatasetFacetProvider;
 import org.apache.flink.connector.kafka.source.KafkaSource;
 
-import com.google.common.reflect.TypeToken;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.serialization.Serializer;
 import org.slf4j.Logger;
@@ -40,6 +39,7 @@
 import javax.annotation.Nullable;
 
 import java.io.Serializable;
+import java.lang.reflect.Type;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
@@ -433,15 +433,17 @@ public Optional<TypeDatasetFacet> getTypeDatasetFacet() {
                                         .getProducedType()));
             } else {
                 // gets type information from serialize method signature
-                TypeToken serializationSchemaType =
-                        TypeToken.of(valueSerializationSchema.getClass());
-                Class parameterType =
-                        serializationSchemaType
-                                .resolveType(SerializationSchema.class.getTypeParameters()[0])
-                                .getRawType();
-                if (parameterType != Object.class) {
+                Type type =
+                        TypeExtractor.getParameterType(
+                                SerializationSchema.class, valueSerializationSchema.getClass(), 0);
+                try {
                     return Optional.of(
-                            new DefaultTypeDatasetFacet(TypeInformation.of(parameterType)));
+                            new DefaultTypeDatasetFacet(TypeExtractor.createTypeInfo(type)));
+                } catch (Exception e) {
+                    LOG.info(
+                            "Could not extract type information from {}",
+                            valueSerializationSchema.getClass(),
+                            e);
                 }
             }
             return Optional.empty();
diff --git a/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/DynamicKafkaRecordSerializationSchema.java b/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/DynamicKafkaRecordSerializationSchema.java
@@ -19,8 +19,8 @@
 
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.api.common.serialization.SerializationSchema;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
 import org.apache.flink.connector.kafka.lineage.DefaultKafkaDatasetFacet;
 import org.apache.flink.connector.kafka.lineage.DefaultKafkaDatasetIdentifier;
 import org.apache.flink.connector.kafka.lineage.DefaultTypeDatasetFacet;
@@ -36,11 +36,13 @@
 import org.apache.flink.types.RowKind;
 import org.apache.flink.util.Preconditions;
 
-import com.google.common.reflect.TypeToken;
 import org.apache.kafka.clients.producer.ProducerRecord;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import javax.annotation.Nullable;
 
+import java.lang.reflect.Type;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -58,6 +60,8 @@ class DynamicKafkaRecordSerializationSchema
         implements KafkaRecordSerializationSchema<RowData>,
                 KafkaDatasetFacetProvider,
                 TypeDatasetFacetProvider {
+    private static final Logger LOG =
+            LoggerFactory.getLogger(DynamicKafkaRecordSerializationSchema.class);
 
     private final Set<String> topics;
     private final Pattern topicPattern;
@@ -208,13 +212,16 @@ public Optional<TypeDatasetFacet> getTypeDatasetFacet() {
                             ((ResultTypeQueryable<?>) this.valueSerialization).getProducedType()));
         } else {
             // gets type information from serialize method signature
-            TypeToken serializationSchemaType = TypeToken.of(valueSerialization.getClass());
-            Class parameterType =
-                    serializationSchemaType
-                            .resolveType(SerializationSchema.class.getTypeParameters()[0])
-                            .getRawType();
-            if (parameterType != Object.class) {
-                return Optional.of(new DefaultTypeDatasetFacet(TypeInformation.of(parameterType)));
+            Type type =
+                    TypeExtractor.getParameterType(
+                            SerializationSchema.class, valueSerialization.getClass(), 0);
+            try {
+                return Optional.of(new DefaultTypeDatasetFacet(TypeExtractor.createTypeInfo(type)));
+            } catch (Exception e) {
+                LOG.info(
+                        "Could not extract type information from {}",
+                        valueSerialization.getClass(),
+                        e);
             }
         }
         return Optional.empty();
diff --git a/pom.xml b/pom.xml
@@ -264,6 +264,8 @@ under the License.
 				<groupId>com.google.guava</groupId>
 				<artifactId>guava</artifactId>
 				<version>${guava.version}</version>
+				<!-- Don't use guava in production code-->
+				<scope>test</scope>
 			</dependency>
 
             <!-- Flink dependencies -->
