From e65dd83abf803fcb0efdcfbe812490a783fa35d6 Mon Sep 17 00:00:00 2001 From: Yanjun Qiu <153984347+qiuyanjun888@users.noreply.github.com> Date: Sun, 5 Jul 2026 12:58:17 +0800 Subject: [PATCH] [FLINK-39429][cdc-base] Avoid shaded Guava in incremental source fetchers --- .../assigner/SnapshotSplitAssigner.java | 8 +- .../IncrementalSourceEnumerator.java | 17 +- .../IncrementalSourceScanFetcher.java | 14 +- .../IncrementalSourceStreamFetcher.java | 8 +- ...ncrementalSourceFetcherDependencyTest.java | 250 ++++++++++++++++++ 5 files changed, 280 insertions(+), 17 deletions(-) create mode 100644 flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/test/java/org/apache/flink/cdc/connectors/base/source/reader/external/IncrementalSourceFetcherDependencyTest.java diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/assigner/SnapshotSplitAssigner.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/assigner/SnapshotSplitAssigner.java index 0b1a4100bc7..4ed81f9507c 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/assigner/SnapshotSplitAssigner.java +++ b/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); diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/enumerator/IncrementalSourceEnumerator.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/enumerator/IncrementalSourceEnumerator.java index a9a8c34db0d..6099c06ad62 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/enumerator/IncrementalSourceEnumerator.java +++ b/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 List> partition(List list, int partitionSize) { + if (partitionSize <= 0) { + throw new IllegalArgumentException("Partition size must be positive"); + } + List> 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; + } } diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/external/IncrementalSourceScanFetcher.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/external/IncrementalSourceScanFetcher.java index 3136bb97d25..4821a9bff6d 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/external/IncrementalSourceScanFetcher.java +++ b/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 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); diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/external/IncrementalSourceStreamFetcher.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/external/IncrementalSourceStreamFetcher.java index 937fefe67a1..660d38dde3d 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/external/IncrementalSourceStreamFetcher.java +++ b/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 { + 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(); diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/test/java/org/apache/flink/cdc/connectors/base/source/reader/external/IncrementalSourceFetcherDependencyTest.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/test/java/org/apache/flink/cdc/connectors/base/source/reader/external/IncrementalSourceFetcherDependencyTest.java new file mode 100644 index 00000000000..677ae111b52 --- /dev/null +++ b/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 classesReferencingFlinkShadedGuava31; + try (Stream 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 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 outputBuffer, SourceRecord changeRecord) {} + + @Override + public List formatMessageTimestamp(Collection 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); + } + } +}