[FLINK-39429][cdc-base] Avoid shaded Guava in incremental source fetchers

flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/assigner/SnapshotSplitAssigner.java

@@ -32,8 +32,6 @@
 import org.apache.flink.util.FlinkRuntimeException;
 import org.apache.flink.util.Preconditions;
 
-import org.apache.flink.shaded.guava31.com.google.common.util.concurrent.ThreadFactoryBuilder;
-
 import io.debezium.relational.TableId;
 import io.debezium.relational.history.TableChanges;
 import org.slf4j.Logger;
@@ -339,7 +337,11 @@ private void startAsynchronouslySplit() {
         if (chunkSplitter.hasNextChunk() || !remainingTables.isEmpty()) {
             if (splittingExecutorService == null) {
                 ThreadFactory threadFactory =
-                        new ThreadFactoryBuilder().setNameFormat("snapshot-splitting").build();
+                        runnable -> {
+                            Thread thread = Executors.defaultThreadFactory().newThread(runnable);
+                            thread.setName("snapshot-splitting");
+                            return thread;
+                        };
                 this.splittingExecutorService = Executors.newSingleThreadExecutor(threadFactory);
             }
             splittingExecutorService.submit(this::splitChunksForRemainingTables);

flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/enumerator/IncrementalSourceEnumerator.java

@@ -43,8 +43,6 @@
 import org.apache.flink.util.CollectionUtil;
 import org.apache.flink.util.FlinkRuntimeException;
 
-import org.apache.flink.shaded.guava31.com.google.common.collect.Lists;
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -311,8 +309,7 @@ private void sendStreamMetaRequestEvent(int subTask, StreamSplitMetaRequestEvent
                         "The assigner offer empty finished split information, this should not happen");
             }
             finishedSnapshotSplitMeta =
-                    Lists.partition(
-                            finishedSnapshotSplitInfos, sourceConfig.getSplitMetaGroupSize());
+                    partition(finishedSnapshotSplitInfos, sourceConfig.getSplitMetaGroupSize());
         }
         final int requestMetaGroupId = requestEvent.getRequestMetaGroupId();
         final int totalFinishedSplitSizeOfReader = requestEvent.getTotalFinishedSplitSize();
@@ -361,4 +358,16 @@ private void handleLatestFinishedSplitNumberRequest(int subtaskId) {
                             splitAssigner.getFinishedSplitInfos().size()));
         }
     }
+
+    private static <T> List<List<T>> partition(List<T> list, int partitionSize) {
+        if (partitionSize <= 0) {
+            throw new IllegalArgumentException("Partition size must be positive");
+        }
+        List<List<T>> partitions =
+                new ArrayList<>((list.size() + partitionSize - 1) / partitionSize);
+        for (int start = 0; start < list.size(); start += partitionSize) {
+            partitions.add(list.subList(start, Math.min(start + partitionSize, list.size())));
+        }
+        return partitions;
+    }
 }

flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/external/IncrementalSourceScanFetcher.java

@@ -23,8 +23,6 @@
 import org.apache.flink.cdc.connectors.base.source.meta.split.SourceSplitBase;
 import org.apache.flink.util.FlinkRuntimeException;
 
-import org.apache.flink.shaded.guava31.com.google.common.util.concurrent.ThreadFactoryBuilder;
-
 import io.debezium.connector.base.ChangeEventQueue;
 import io.debezium.pipeline.DataChangeEvent;
 import org.apache.kafka.connect.data.Struct;
@@ -75,11 +73,13 @@ public class IncrementalSourceScanFetcher implements Fetcher<SourceRecords, Sour
     public IncrementalSourceScanFetcher(FetchTask.Context taskContext, int subtaskId) {
         this.taskContext = taskContext;
         ThreadFactory threadFactory =
-                new ThreadFactoryBuilder()
-                        .setNameFormat("debezium-snapshot-reader-" + subtaskId)
-                        .setUncaughtExceptionHandler(
-                                (thread, throwable) -> setReadException(throwable))
-                        .build();
+                runnable -> {
+                    Thread thread = Executors.defaultThreadFactory().newThread(runnable);
+                    thread.setName("debezium-snapshot-reader-" + subtaskId);
+                    thread.setUncaughtExceptionHandler(
+                            (caughtThread, throwable) -> setReadException(throwable));
+                    return thread;
+                };
         this.executorService = Executors.newSingleThreadExecutor(threadFactory);
         this.hasNextElement = new AtomicBoolean(false);
         this.reachEnd = new AtomicBoolean(false);

flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/external/IncrementalSourceStreamFetcher.java

@@ -25,8 +25,6 @@
 import org.apache.flink.cdc.connectors.base.utils.SplitKeyUtils;
 import org.apache.flink.util.FlinkRuntimeException;
 
-import org.apache.flink.shaded.guava31.com.google.common.util.concurrent.ThreadFactoryBuilder;
-
 import io.debezium.connector.base.ChangeEventQueue;
 import io.debezium.pipeline.DataChangeEvent;
 import io.debezium.relational.TableId;
@@ -76,7 +74,11 @@ public class IncrementalSourceStreamFetcher implements Fetcher<SourceRecords, So
     public IncrementalSourceStreamFetcher(FetchTask.Context taskContext, int subTaskId) {
         this.taskContext = taskContext;
         ThreadFactory threadFactory =
-                new ThreadFactoryBuilder().setNameFormat("debezium-reader-" + subTaskId).build();
+                runnable -> {
+                    Thread thread = Executors.defaultThreadFactory().newThread(runnable);
+                    thread.setName("debezium-reader-" + subTaskId);
+                    return thread;
+                };
         this.executorService = Executors.newSingleThreadExecutor(threadFactory);
         this.pureStreamPhaseTables = new HashSet<>();
         this.isBackfillSkipped = taskContext.getSourceConfig().isSkipSnapshotBackfill();

flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/test/java/org/apache/flink/cdc/connectors/base/source/reader/external/IncrementalSourceFetcherDependencyTest.java

@@ -0,0 +1,250 @@
+/*
+ * 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.flink.cdc.connectors.base.source.reader.external;
+
+import org.apache.flink.cdc.connectors.base.config.SourceConfig;
+import org.apache.flink.cdc.connectors.base.dialect.DataSourceDialect;
+import org.apache.flink.cdc.connectors.base.options.StartupOptions;
+import org.apache.flink.cdc.connectors.base.source.meta.offset.Offset;
+import org.apache.flink.cdc.connectors.base.source.meta.split.SourceSplitBase;
+
+import io.debezium.connector.base.ChangeEventQueue;
+import io.debezium.pipeline.DataChangeEvent;
+import io.debezium.relational.TableId;
+import io.debezium.relational.Tables;
+import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.source.SourceRecord;
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.Constructor;
+import java.lang.reflect.Method;
+import java.net.URL;
+import java.net.URLClassLoader;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+
+/** Tests that incremental source fetchers do not depend on Flink's Guava 31 shade. */
+class IncrementalSourceFetcherDependencyTest {
+
+    private static final String SHADED_GUAVA31_PREFIX = "org.apache.flink.shaded.guava31.";
+
+    @Test
+    void incrementalFetchersCanBeConstructedWithoutFlinkShadedGuava31() throws Exception {
+        FetchTask.Context context = new TestingFetchTaskContext();
+
+        URL classesUrl = Paths.get("target/classes").toAbsolutePath().toUri().toURL();
+        try (Guava31BlockingClassLoader classLoader =
+                new Guava31BlockingClassLoader(
+                        new URL[] {classesUrl}, getClass().getClassLoader())) {
+            assertThatCode(
+                            () ->
+                                    constructAndCloseFetcher(
+                                            classLoader,
+                                            "org.apache.flink.cdc.connectors.base.source.reader.external.IncrementalSourceScanFetcher",
+                                            context))
+                    .doesNotThrowAnyException();
+            assertThatCode(
+                            () ->
+                                    constructAndCloseFetcher(
+                                            classLoader,
+                                            "org.apache.flink.cdc.connectors.base.source.reader.external.IncrementalSourceStreamFetcher",
+                                            context))
+                    .doesNotThrowAnyException();
+        }
+    }
+
+    @Test
+    void productionClassesDoNotReferenceFlinkShadedGuava31() throws Exception {
+        List<Path> classesReferencingFlinkShadedGuava31;
+        try (Stream<Path> classes = Files.walk(Paths.get("target/classes"))) {
+            classesReferencingFlinkShadedGuava31 =
+                    classes.filter(path -> path.toString().endsWith(".class"))
+                            .filter(this::containsFlinkShadedGuava31Reference)
+                            .collect(Collectors.toList());
+        }
+
+        assertThat(classesReferencingFlinkShadedGuava31)
+                .as("Production classes should not depend on Flink's Guava 31 shade")
+                .isEmpty();
+    }
+
+    private boolean containsFlinkShadedGuava31Reference(Path path) {
+        try {
+            return new String(Files.readAllBytes(path), StandardCharsets.ISO_8859_1)
+                    .contains("org/apache/flink/shaded/guava31");
+        } catch (Exception e) {
+            throw new AssertionError("Failed to inspect class file " + path, e);
+        }
+    }
+
+    private static void constructAndCloseFetcher(
+            ClassLoader classLoader, String className, FetchTask.Context context) throws Exception {
+        Class<?> fetcherClass = Class.forName(className, true, classLoader);
+        Constructor<?> constructor =
+                fetcherClass.getConstructor(FetchTask.Context.class, int.class);
+        Object fetcher = constructor.newInstance(context, 0);
+        Method close = fetcherClass.getMethod("close");
+        close.invoke(fetcher);
+    }
+
+    private static final class TestingFetchTaskContext implements FetchTask.Context {
+
+        private final SourceConfig sourceConfig = new TestingSourceConfig();
+
+        @Override
+        public void configure(SourceSplitBase sourceSplitBase) {}
+
+        @Override
+        public ChangeEventQueue<DataChangeEvent> getQueue() {
+            return null;
+        }
+
+        @Override
+        public TableId getTableId(SourceRecord record) {
+            return null;
+        }
+
+        @Override
+        public Tables.TableFilter getTableFilter() {
+            return null;
+        }
+
+        @Override
+        public Offset getStreamOffset(SourceRecord record) {
+            return null;
+        }
+
+        @Override
+        public boolean isDataChangeRecord(SourceRecord record) {
+            return false;
+        }
+
+        @Override
+        public boolean isRecordBetween(
+                SourceRecord record, Object[] splitStart, Object[] splitEnd) {
+            return false;
+        }
+
+        @Override
+        public void rewriteOutputBuffer(
+                Map<Struct, SourceRecord> outputBuffer, SourceRecord changeRecord) {}
+
+        @Override
+        public List<SourceRecord> formatMessageTimestamp(Collection<SourceRecord> snapshotRecords) {
+            return List.copyOf(snapshotRecords);
+        }
+
+        @Override
+        public void close() {}
+
+        @Override
+        public DataSourceDialect getDataSourceDialect() {
+            return null;
+        }
+
+        @Override
+        public SourceConfig getSourceConfig() {
+            return sourceConfig;
+        }
+    }
+
+    private static final class TestingSourceConfig implements SourceConfig {
+
+        @Override
+        public StartupOptions getStartupOptions() {
+            return StartupOptions.initial();
+        }
+
+        @Override
+        public int getSplitSize() {
+            return 0;
+        }
+
+        @Override
+        public int getSplitMetaGroupSize() {
+            return 0;
+        }
+
+        @Override
+        public boolean isIncludeSchemaChanges() {
+            return false;
+        }
+
+        @Override
+        public boolean isCloseIdleReaders() {
+            return false;
+        }
+
+        @Override
+        public boolean isSkipSnapshotBackfill() {
+            return false;
+        }
+
+        @Override
+        public boolean isScanNewlyAddedTableEnabled() {
+            return false;
+        }
+
+        @Override
+        public boolean isAssignUnboundedChunkFirst() {
+            return false;
+        }
+    }
+
+    private static class Guava31BlockingClassLoader extends URLClassLoader {
+
+        private Guava31BlockingClassLoader(URL[] urls, ClassLoader parent) {
+            super(urls, parent);
+        }
+
+        @Override
+        protected Class<?> loadClass(String name, boolean resolve) throws ClassNotFoundException {
+            if (name.startsWith(SHADED_GUAVA31_PREFIX)) {
+                throw new ClassNotFoundException(name);
+            }
+            if (name.endsWith("IncrementalSourceScanFetcher")
+                    || name.endsWith("IncrementalSourceStreamFetcher")) {
+                synchronized (getClassLoadingLock(name)) {
+                    Class<?> loadedClass = findLoadedClass(name);
+                    if (loadedClass == null) {
+                        try {
+                            loadedClass = findClass(name);
+                        } catch (ClassNotFoundException ignored) {
+                            loadedClass = super.loadClass(name, false);
+                        }
+                    }
+                    if (resolve) {
+                        resolveClass(loadedClass);
+                    }
+                    return loadedClass;
+                }
+            }
+            return super.loadClass(name, resolve);
+        }
+    }
+}