This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 432d4cfcd5 [Fix][CI] Fix missed detections and false positives in
seatunnel-ci-tools checks (#12676)
432d4cfcd5 is described below
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 {}
