This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12676-3b522a90dc534260930c25c61d9010fae7538e55 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 432d4cfcd56d94e0091daad4109733b7c8a81590 Author: Jast <[email protected]> AuthorDate: Thu Oct 15 12:26:46 2026 +0000 [Fix][CI] Fix missed detections and false positives in seatunnel-ci-tools checks (#12676) Co-authored-by: root <[email protected]> Co-authored-by: jast <[email protected]> --- .../seatunnel/api/ConnectorOptionCheckTest.java | 179 +++++++++++--------- .../apache/seatunnel/api/ImportClassCheckTest.java | 36 ++-- .../seatunnel/api/SerialVersionUIDCheckerTest.java | 181 +++++++++++++++++---- .../source/OceanBaseIncrementalSourceFactory.java | 12 +- .../source/OceanBaseIncrementalSourceOptions.java | 35 ++++ .../OpengaussIncrementalSourceOptions.java | 29 ++++ 6 files changed, 343 insertions(+), 129 deletions(-) diff --git a/seatunnel-ci-tools/src/test/java/org/apache/seatunnel/api/ConnectorOptionCheckTest.java b/seatunnel-ci-tools/src/test/java/org/apache/seatunnel/api/ConnectorOptionCheckTest.java index 28484aa08e..20ff292bcb 100644 --- a/seatunnel-ci-tools/src/test/java/org/apache/seatunnel/api/ConnectorOptionCheckTest.java +++ b/seatunnel-ci-tools/src/test/java/org/apache/seatunnel/api/ConnectorOptionCheckTest.java @@ -23,9 +23,11 @@ import org.junit.jupiter.api.Test; import com.github.javaparser.JavaParser; import com.github.javaparser.ParseResult; import com.github.javaparser.ast.CompilationUnit; -import com.github.javaparser.ast.NodeList; import com.github.javaparser.ast.body.ClassOrInterfaceDeclaration; -import com.github.javaparser.ast.type.ClassOrInterfaceType; +import com.github.javaparser.symbolsolver.JavaSymbolSolver; +import com.github.javaparser.symbolsolver.resolution.typesolvers.CombinedTypeSolver; +import com.github.javaparser.symbolsolver.resolution.typesolvers.JavaParserTypeSolver; +import com.github.javaparser.symbolsolver.resolution.typesolvers.ReflectionTypeSolver; import lombok.extern.slf4j.Slf4j; import java.io.File; @@ -34,9 +36,11 @@ import java.nio.file.FileVisitOption; import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.Paths; +import java.util.Arrays; import java.util.HashSet; import java.util.List; import java.util.Set; +import java.util.TreeSet; import java.util.stream.Collectors; import java.util.stream.Stream; @@ -47,11 +51,56 @@ public class ConnectorOptionCheckTest { "src" + File.separator + "main" + File.separator + "java"; private static final String JAVA_FILE_EXTENSION = ".java"; private static final String CONNECTOR_DIR = "seatunnel-connectors-v2"; - private static final JavaParser JAVA_PARSER = new JavaParser(); + private static final Path SOURCE_ROOT_FRAGMENT = Paths.get("src", "main", "java"); + private static final String SOURCE_FQN = "org.apache.seatunnel.api.source.SeaTunnelSource"; + private static final String SINK_FQN = "org.apache.seatunnel.api.sink.SeaTunnelSink"; + + /** + * Directly known connector base classes. Only used as the fallback when the type hierarchy of a + * class cannot be resolved; the resolved check covers these and every other transitive base + * class automatically. + */ + private static final Set<String> DIRECT_CONNECTOR_BASE_CLASSES = + new HashSet<>( + Arrays.asList( + "AbstractSimpleSink", + "AbstractSingleSplitSource", + "IncrementalSource", + "BaseMultipleTableFileSink", + "BaseFileSource", + "BaseFileSink", + "HttpSource", + "HttpSink")); + + private static final JavaParser JAVA_PARSER; + + static { + CombinedTypeSolver typeSolver = new CombinedTypeSolver(); + typeSolver.add(new ReflectionTypeSolver()); + try (Stream<Path> paths = Files.walk(Paths.get(".."), FileVisitOption.FOLLOW_LINKS)) { + // Only real source roots can resolve symbols; registering nested directories or + // individual files only slows every symbol resolution down. + paths.filter(Files::isDirectory) + .filter(path -> path.endsWith(SOURCE_ROOT_FRAGMENT)) + .forEach( + path -> { + try { + typeSolver.add(new JavaParserTypeSolver(path.toFile())); + } catch (Exception e) { + // ignore + } + }); + } catch (IOException e) { + log.error("Failed to setup type solver", e); + } + JAVA_PARSER = new JavaParser(); + JAVA_PARSER.getParserConfiguration().setSymbolResolver(new JavaSymbolSolver(typeSolver)); + } @Test public void checkConnectorOptionExist() { - Set<String> connectorOptionFileNames = new HashSet<>(); + // A TreeSet keeps the failure output stable across runs. + Set<String> connectorOptionFileNames = new TreeSet<>(); try (Stream<Path> paths = Files.walk(Paths.get(".."), FileVisitOption.FOLLOW_LINKS)) { List<Path> connectorClassPaths = paths.filter( @@ -80,88 +129,18 @@ public class ConnectorOptionCheckTest { || classDeclaration.isInterface()) { continue; } - NodeList<ClassOrInterfaceType> - implementedTypes = - classDeclaration - .getImplementedTypes(); - implementedTypes.forEach( - implementedType -> { - if (implementedType - .getNameAsString() - .equals( - "SeaTunnelSource") - || implementedType - .getNameAsString() - .equals( - "SeaTunnelSink")) { - connectorOptionFileNames.add( - path.getFileName() - .toString() - .replace( - JAVA_FILE_EXTENSION, - "") - .concat( - "Options")); - } - }); - NodeList<ClassOrInterfaceType> extendedTypes = - classDeclaration.getExtendedTypes(); - extendedTypes.forEach( - extendedType -> { - if (extendedType - .getNameAsString() - .equals( - "AbstractSimpleSink") - || extendedType - .getNameAsString() - .equals( - "AbstractSingleSplitSource") - || extendedType - .getNameAsString() - .equals( - "IncrementalSource") - || extendedType - .getNameAsString() - .equals( - "BaseMultipleTableFileSink") - || extendedType - .getNameAsString() - .equals( - "BaseFileSource") - || extendedType - .getNameAsString() - .equals( - "BaseFileSink") - || extendedType - .getNameAsString() - .equals( - "HttpSource") - || extendedType - .getNameAsString() - .equals( - "HttpSink")) { - connectorOptionFileNames.add( - path.getFileName() - .toString() - .replace( - JAVA_FILE_EXTENSION, - "") - .concat( - "Options")); - } - }); + if (isSeaTunnelConnector(classDeclaration)) { + connectorOptionFileNames.add( + classNameOf(path) + .concat("Options")); + } } }); } catch (IOException e) { throw new RuntimeException(e); } }); - connectorClassPaths.forEach( - path -> { - String className = - path.getFileName().toString().replace(JAVA_FILE_EXTENSION, ""); - connectorOptionFileNames.remove(className); - }); + connectorClassPaths.forEach(path -> connectorOptionFileNames.remove(classNameOf(path))); Assertions.assertEquals( 0, @@ -176,4 +155,44 @@ public class ConnectorOptionCheckTest { throw new RuntimeException(e); } } + + private String classNameOf(Path path) { + String fileName = path.getFileName().toString(); + return fileName.endsWith(JAVA_FILE_EXTENSION) + ? fileName.substring(0, fileName.length() - JAVA_FILE_EXTENSION.length()) + : fileName; + } + + private boolean isSeaTunnelConnector(ClassOrInterfaceDeclaration classDeclaration) { + try { + // Resolve the whole hierarchy instead of matching only the directly declared + // types: connectors that implement SeaTunnelSource/SeaTunnelSink through an + // intermediate base class were missed by the direct check. + return classDeclaration.resolve().getAllAncestors().stream() + .anyMatch( + ancestor -> { + String name = ancestor.getQualifiedName(); + return SOURCE_FQN.equals(name) || SINK_FQN.equals(name); + }); + } catch (Exception e) { + // Fall back to direct-name matching when the hierarchy cannot be resolved, for + // example when an ancestor comes from a dependency that is not a source root. + return matchesConnectorByDirectTypes(classDeclaration); + } + } + + private boolean matchesConnectorByDirectTypes(ClassOrInterfaceDeclaration classDeclaration) { + return classDeclaration.getImplementedTypes().stream() + .anyMatch( + type -> { + String name = type.getNameAsString(); + return name.equals("SeaTunnelSource") + || name.equals("SeaTunnelSink"); + }) + || classDeclaration.getExtendedTypes().stream() + .anyMatch( + type -> + DIRECT_CONNECTOR_BASE_CLASSES.contains( + type.getNameAsString())); + } } diff --git a/seatunnel-ci-tools/src/test/java/org/apache/seatunnel/api/ImportClassCheckTest.java b/seatunnel-ci-tools/src/test/java/org/apache/seatunnel/api/ImportClassCheckTest.java index b15e5ac8ce..e12ee6014d 100644 --- a/seatunnel-ci-tools/src/test/java/org/apache/seatunnel/api/ImportClassCheckTest.java +++ b/seatunnel-ci-tools/src/test/java/org/apache/seatunnel/api/ImportClassCheckTest.java @@ -93,25 +93,17 @@ public class ImportClassCheckTest { @Test public void commonLang2Check() { - // both common-lang and common-lang3 share the same prefix org.apache.commons.lang - Map<String, List<String>> commonLangMap = - checkImportClassPrefix( - Arrays.asList("org.apache.commons.lang"), - Collections.emptyList(), - Collections.emptyList()); - // common-lang3 - Map<String, List<String>> commonLang3Map = + // Both commons-lang2 (org.apache.commons.lang) and commons-lang3 + // (org.apache.commons.lang3) match the org.apache.commons.lang prefix, so lang3 + // imports must be excluded per import statement. Filtering whole files instead + // would let a file that mixes lang2 and lang3 imports pass unnoticed. + Map<String, List<String>> errorMap = checkImportClassPrefix( - Arrays.asList("org.apache.commons.lang3"), + Collections.singletonList("org.apache.commons.lang"), + Collections.singletonList("org.apache.commons.lang3"), Collections.emptyList(), Collections.emptyList()); - // find the one in common-lang but not common-lang3 - Map<String, List<String>> errorMap = - commonLangMap.entrySet().stream() - .filter(entry -> !commonLang3Map.containsKey(entry.getKey())) - .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); - Assertions.assertEquals( 0, errorMap.size(), shadeErrorMsg("org.apache.commons.lang", errorMap)); log.info("check org.apache.commons.lang successfully"); @@ -205,6 +197,15 @@ public class ImportClassCheckTest { private Map<String, List<String>> checkImportClassPrefix( List<String> prefixList, List<String> packageCheckList, List<String> packageWhiteList) { + return checkImportClassPrefix( + prefixList, Collections.emptyList(), packageCheckList, packageWhiteList); + } + + private Map<String, List<String>> checkImportClassPrefix( + List<String> prefixList, + List<String> excludePrefixList, + List<String> packageCheckList, + List<String> packageWhiteList) { List<String> pathWhiteList = packageWhiteList.stream() .map(whitePackage -> whitePackage.replace(".", isWindows ? "\\" : "/")) @@ -233,7 +234,10 @@ public class ImportClassCheckTest { String importClz = importMetadata.getClassName(); return prefixList.stream() - .anyMatch(importClz::startsWith); + .anyMatch(importClz::startsWith) + && excludePrefixList.stream() + .noneMatch( + importClz::startsWith); }) .map(this::getImportClassLineNum) .collect(Collectors.toList()); diff --git a/seatunnel-ci-tools/src/test/java/org/apache/seatunnel/api/SerialVersionUIDCheckerTest.java b/seatunnel-ci-tools/src/test/java/org/apache/seatunnel/api/SerialVersionUIDCheckerTest.java index dc13a1d9b6..5650e55207 100644 --- a/seatunnel-ci-tools/src/test/java/org/apache/seatunnel/api/SerialVersionUIDCheckerTest.java +++ b/seatunnel-ci-tools/src/test/java/org/apache/seatunnel/api/SerialVersionUIDCheckerTest.java @@ -28,7 +28,11 @@ import org.slf4j.LoggerFactory; import com.github.javaparser.JavaParser; import com.github.javaparser.ParseResult; import com.github.javaparser.ast.CompilationUnit; +import com.github.javaparser.ast.body.BodyDeclaration; import com.github.javaparser.ast.body.ClassOrInterfaceDeclaration; +import com.github.javaparser.ast.body.FieldDeclaration; +import com.github.javaparser.ast.expr.Expression; +import com.github.javaparser.ast.expr.UnaryExpr; import com.github.javaparser.ast.type.ClassOrInterfaceType; import com.github.javaparser.ast.type.Type; import com.github.javaparser.resolution.declarations.ResolvedReferenceTypeDeclaration; @@ -50,6 +54,7 @@ import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.Set; import java.util.stream.Collectors; import java.util.stream.Stream; @@ -125,9 +130,23 @@ public class SerialVersionUIDCheckerTest { + File.separator + "job"); private static final JavaParser JAVA_PARSER; - private static final Set<String> checkedClasses = new HashSet<>(); + // Classes visited by the type-argument path, which only asserts that a serialVersionUID + // field exists. Tracked separately from fullyValidatedClasses so that the governed-class + // path (checkClassDeclaration) still runs its full "static final long, not -1L" validation + // on these classes: sharing one set would let whichever path visits a class first (for + // example JdbcSourceConfig, both a *Config class and a generic state argument of + // IncrementalSource subclasses) short-circuit the other, and the visit order depends on + // Files.walk, i.e. on the filesystem. + private static final Set<String> existenceCheckedClasses = new HashSet<>(); + // Classes that went through the full declaration validation in checkClassDeclaration. + // The existence-only path skips these because the full validation subsumes it. + private static final Set<String> fullyValidatedClasses = new HashSet<>(); private static final Map<String, ClassOrInterfaceDeclaration> classDeclarationMap = new HashMap<>(); + private static final Path SOURCE_ROOT_FRAGMENT = Paths.get("src", "main", "java"); + private static final String SOURCE_FQN = "org.apache.seatunnel.api.source.SeaTunnelSource"; + private static final String SINK_FQN = "org.apache.seatunnel.api.sink.SeaTunnelSink"; + private static final String SERIAL_VERSION_UID_FIELD = "serialVersionUID"; static { CombinedTypeSolver typeSolver = new CombinedTypeSolver(); @@ -140,7 +159,12 @@ public class SerialVersionUIDCheckerTest { private static void setupTypeSolver(CombinedTypeSolver typeSolver) { try (Stream<Path> paths = Files.walk(Paths.get(".."), FileVisitOption.FOLLOW_LINKS)) { - paths.filter(path -> path.toString().contains("src/main/java")) + // Only real source roots can resolve symbols. Registering nested directories or + // individual files creates thousands of solvers that never match anything and + // slow every single symbol resolution down. Paths#get builds the fragment with + // the platform separator, so this also works on Windows. + paths.filter(Files::isDirectory) + .filter(path -> path.endsWith(SOURCE_ROOT_FRAGMENT)) .forEach( path -> { try { @@ -342,6 +366,25 @@ public class SerialVersionUIDCheckerTest { } private boolean implementsSeaTunnelSourceOrSink(ClassOrInterfaceDeclaration classDeclaration) { + try { + // Resolve the full hierarchy instead of only the directly declared types: most + // connectors implement SeaTunnelSource/SeaTunnelSink through an intermediate base + // class such as IncrementalSource or HttpSource, which the direct check missed. + return classDeclaration.resolve().getAllAncestors().stream() + .anyMatch( + ancestor -> { + String name = ancestor.getQualifiedName(); + return SOURCE_FQN.equals(name) || SINK_FQN.equals(name); + }); + } catch (Exception e) { + // Fall back to the direct-name check when the hierarchy cannot be resolved, for + // example when an ancestor comes from a dependency that is not a source root. + return matchesSeaTunnelSourceOrSinkDirectly(classDeclaration); + } + } + + private boolean matchesSeaTunnelSourceOrSinkDirectly( + ClassOrInterfaceDeclaration classDeclaration) { return classDeclaration.getImplementedTypes().stream() .anyMatch( type -> { @@ -353,23 +396,26 @@ public class SerialVersionUIDCheckerTest { private void checkImplementedTypes( ClassOrInterfaceDeclaration classDeclaration, List<String> missingSerialVersionUID) { - classDeclaration - .getImplementedTypes() + // The connector's state/config types can be bound on the implemented types + // (implements SeaTunnelSource<...>) or on the extended ones + // (extends IncrementalSource<...>), so both have to be inspected. + Stream.concat( + classDeclaration.getImplementedTypes().stream(), + classDeclaration.getExtendedTypes().stream()) .forEach( - implementedType -> { - implementedType - .getTypeArguments() - .ifPresent( - typeArgs -> { - for (Type typeArg : typeArgs) { - if (typeArg.isClassOrInterfaceType()) { - checkClassType( - typeArg.asClassOrInterfaceType(), - missingSerialVersionUID); + type -> + type.getTypeArguments() + .ifPresent( + typeArgs -> { + for (Type typeArg : typeArgs) { + if (typeArg.isClassOrInterfaceType()) { + checkClassType( + typeArg + .asClassOrInterfaceType(), + missingSerialVersionUID); + } } - } - }); - }); + })); } private void checkClassType( @@ -387,10 +433,11 @@ public class SerialVersionUIDCheckerTest { return; } String paramTypeName = typeDeclaration.getQualifiedName(); - if (!checkedClasses.contains(paramTypeName)) { + if (!existenceCheckedClasses.contains(paramTypeName) + && !fullyValidatedClasses.contains(paramTypeName)) { // Check if the class is abstract and return early if it is if (isAbstractClass(typeDeclaration)) { - checkedClasses.add(paramTypeName); + existenceCheckedClasses.add(paramTypeName); return; } @@ -398,7 +445,7 @@ public class SerialVersionUIDCheckerTest { missingSerialVersionUID.add(paramTypeName); LOG.warn("Class {} is missing serialVersionUID field", paramTypeName); } - checkedClasses.add(paramTypeName); + existenceCheckedClasses.add(paramTypeName); } } } catch (Exception e) { @@ -411,12 +458,13 @@ public class SerialVersionUIDCheckerTest { try { ResolvedReferenceTypeDeclaration typeDeclaration = classDeclaration.resolve(); String className = typeDeclaration.getQualifiedName(); - if (!checkedClasses.contains(className)) { - if (!hasSerialVersionUID(typeDeclaration)) { - missingSerialVersionUID.add(className); - LOG.warn("Class {} is missing serialVersionUID field", className); + if (!fullyValidatedClasses.contains(className)) { + String problem = checkSerialVersionUIDDeclaration(classDeclaration); + if (problem != null) { + missingSerialVersionUID.add(className + " - " + problem); + LOG.warn("Class {} has an invalid serialVersionUID: {}", className, problem); } - checkedClasses.add(className); + fullyValidatedClasses.add(className); } } catch (Exception e) { LOG.warn( @@ -424,6 +472,72 @@ public class SerialVersionUIDCheckerTest { } } + /** + * Validates the serialVersionUID field of the given declaration and returns a description of + * the problem, or null when the field is valid. The field must exist, be declared as {@code + * static final long}, and must not use the {@code -1L} sentinel value, matching what the + * failure message of this checker has always been asking for. + */ + private String checkSerialVersionUIDDeclaration(ClassOrInterfaceDeclaration classDeclaration) { + Optional<FieldDeclaration> field = + classDeclaration.getMembers().stream() + .filter(BodyDeclaration::isFieldDeclaration) + .map(BodyDeclaration::asFieldDeclaration) + .filter( + f -> + f.getVariables().stream() + .anyMatch( + v -> + SERIAL_VERSION_UID_FIELD.equals( + v.getNameAsString()))) + .findFirst(); + if (!field.isPresent()) { + return "missing serialVersionUID field"; + } + FieldDeclaration fieldDeclaration = field.get(); + if (!fieldDeclaration.isStatic() + || !fieldDeclaration.isFinal() + || fieldDeclaration.getVariables().size() != 1 + || !fieldDeclaration.getVariables().get(0).getType().asString().equals("long")) { + return "serialVersionUID must be declared as `private static final long serialVersionUID`"; + } + Long value = + longLiteralValue( + fieldDeclaration.getVariables().get(0).getInitializer().orElse(null)); + if (value != null && value == -1L) { + return "serialVersionUID must not be -1L, it must be a fixed value so that" + + " serialized job graphs and states stay compatible across releases"; + } + return null; + } + + /** + * Returns the value of a plain long literal such as {@code 1L} or {@code -1L}, or null when the + * initializer is missing or is not a plain literal. + */ + private Long longLiteralValue(Expression expression) { + if (expression == null) { + return null; + } + boolean negated = false; + Expression current = expression; + if (current.isUnaryExpr() + && current.asUnaryExpr().getOperator() == UnaryExpr.Operator.MINUS) { + negated = true; + current = current.asUnaryExpr().getExpression(); + } + if (!current.isLiteralStringValueExpr()) { + return null; + } + String literal = current.asLiteralStringValueExpr().getValue().replaceFirst("[lL]$", ""); + try { + long value = Long.parseLong(literal); + return negated ? -value : value; + } catch (NumberFormatException e) { + return null; + } + } + private boolean isSerializable(ResolvedReferenceType resolvedType) { return resolvedType.getQualifiedName().equals("java.io.Serializable") || resolvedType.getAllAncestors().stream() @@ -464,14 +578,24 @@ public class SerialVersionUIDCheckerTest { return classDeclaration.isAbstract(); } - return false; + // The map only holds declarations from the checked paths, so fall back to the AST of + // the resolved declaration for classes anywhere else in the sources. Treat anything + // we still cannot inspect (e.g. types resolved from reflection) as abstract: skipping + // them is safe, while assuming they are concrete could produce false positives. + return typeDeclaration + .toAst() + .map( + node -> + node instanceof ClassOrInterfaceDeclaration + && ((ClassOrInterfaceDeclaration) node).isAbstract()) + .orElse(true); } private String generateErrorMessage(List<String> missingSerialVersionUID) { StringBuilder errorMessage = new StringBuilder(); errorMessage.append("=================================================================\n"); errorMessage.append( - "Test failed: The following classes are missing serialVersionUID fields\n"); + "Test failed: The following classes have an invalid or missing serialVersionUID field\n"); errorMessage.append("=================================================================\n"); errorMessage .append("A total of ") @@ -509,7 +633,8 @@ public class SerialVersionUIDCheckerTest { @AfterAll public static void cleanup() { - checkedClasses.clear(); + existenceCheckedClasses.clear(); + fullyValidatedClasses.clear(); classDeclarationMap.clear(); } } diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oceanbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oceanbase/source/OceanBaseIncrementalSourceFactory.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oceanbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oceanbase/source/OceanBaseIncrementalSourceFactory.java index 3c5599d90e..046e66ef7d 100644 --- a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oceanbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oceanbase/source/OceanBaseIncrementalSourceFactory.java +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oceanbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oceanbase/source/OceanBaseIncrementalSourceFactory.java @@ -45,8 +45,6 @@ import java.util.Map; @AutoService(Factory.class) public class OceanBaseIncrementalSourceFactory extends MySqlIncrementalSourceFactory { - private static final String MYSQL_COMPATIBLE_MODE = "mysql"; - /** * Return the identifier used in SeaTunnel source config. * @@ -110,14 +108,18 @@ public class OceanBaseIncrementalSourceFactory extends MySqlIncrementalSourceFac */ ReadonlyConfig mysqlCompatibleConfig(ReadonlyConfig config) { String compatibleMode = - config.getOptional(JdbcCommonOptions.COMPATIBLE_MODE).orElse(MYSQL_COMPATIBLE_MODE); - if (!MYSQL_COMPATIBLE_MODE.equalsIgnoreCase(compatibleMode)) { + config.getOptional(JdbcCommonOptions.COMPATIBLE_MODE) + .orElse(OceanBaseIncrementalSourceOptions.MYSQL_COMPATIBLE_MODE); + if (!OceanBaseIncrementalSourceOptions.MYSQL_COMPATIBLE_MODE.equalsIgnoreCase( + compatibleMode)) { throw new IllegalArgumentException( "OceanBase-CDC supports only MySQL compatible mode, but compatible_mode is " + compatibleMode); } Map<String, Object> options = new HashMap<>(config.getSourceMap()); - options.put(JdbcCommonOptions.COMPATIBLE_MODE.key(), MYSQL_COMPATIBLE_MODE); + options.put( + JdbcCommonOptions.COMPATIBLE_MODE.key(), + OceanBaseIncrementalSourceOptions.MYSQL_COMPATIBLE_MODE); return ReadonlyConfig.fromMap(options); } } diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oceanbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oceanbase/source/OceanBaseIncrementalSourceOptions.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oceanbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oceanbase/source/OceanBaseIncrementalSourceOptions.java new file mode 100644 index 0000000000..488eab4b24 --- /dev/null +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oceanbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oceanbase/source/OceanBaseIncrementalSourceOptions.java @@ -0,0 +1,35 @@ +/* + * 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.seatunnel.connectors.seatunnel.cdc.oceanbase.source; + +import org.apache.seatunnel.connectors.seatunnel.cdc.mysql.config.MySqlIncrementalSourceOptions; + +/** + * Options of the {@link OceanBaseIncrementalSource}. + * + * <p>The connector reuses the MySQL CDC runtime, so it inherits the MySQL CDC options and only adds + * the OceanBase-specific constants on top. + */ +public class OceanBaseIncrementalSourceOptions extends MySqlIncrementalSourceOptions { + + /** + * The only compatible mode supported by the OceanBase CDC source. The source reuses the MySQL + * CDC runtime, so the mode is fixed to MySQL compatible mode for now. + */ + public static final String MYSQL_COMPATIBLE_MODE = "mysql"; +} diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-opengauss/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/opengauss/OpengaussIncrementalSourceOptions.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-opengauss/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/opengauss/OpengaussIncrementalSourceOptions.java new file mode 100644 index 0000000000..eceb0659c9 --- /dev/null +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-opengauss/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/opengauss/OpengaussIncrementalSourceOptions.java @@ -0,0 +1,29 @@ +/* + * 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.seatunnel.connectors.seatunnel.cdc.opengauss; + +import org.apache.seatunnel.connectors.seatunnel.cdc.postgres.config.PostgresIncrementalSourceOptions; + +/** + * Options of the {@link OpengaussIncrementalSource}. + * + * <p>The connector reuses the PostgreSQL CDC runtime through the PG base, so it inherits the + * PostgreSQL CDC options. OpenGauss-specific option surface (for example the startup modes this + * connector accepts) is owned by {@link OpengaussSourceOptions}. + */ +public class OpengaussIncrementalSourceOptions extends PostgresIncrementalSourceOptions {}
