This is an automated email from the ASF dual-hosted git repository. dockerzhang pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/inlong.git
The following commit(s) were added to refs/heads/master by this push: new 6c1494eb30 [INLONG-8062][Sort] Add PostgreSQL source connector on flink 1.15 (#8193) 6c1494eb30 is described below commit 6c1494eb308066207658d5da5e758a96bbc38509 Author: Sting <zpen...@connect.ust.hk> AuthorDate: Mon Jun 12 17:46:05 2023 +0800 [INLONG-8062][Sort] Add PostgreSQL source connector on flink 1.15 (#8193) --- .../src/main/assemblies/sort-connectors-v1.15.xml | 2 +- inlong-sort/sort-flink/sort-flink-v1.15/pom.xml | 14 +- .../sort-flink-v1.15/sort-connectors/pom.xml | 12 + .../sort-connectors/postgres-cdc/pom.xml | 150 +++++ .../sort/postgre/PostgreSQLTableFactory.java | 181 ++++++ .../org.apache.flink.table.factories.Factory | 16 + licenses/inlong-sort-connectors/LICENSE | 699 +++++++++++---------- 7 files changed, 723 insertions(+), 351 deletions(-) diff --git a/inlong-distribution/src/main/assemblies/sort-connectors-v1.15.xml b/inlong-distribution/src/main/assemblies/sort-connectors-v1.15.xml index d5c3165050..5ef95c3243 100644 --- a/inlong-distribution/src/main/assemblies/sort-connectors-v1.15.xml +++ b/inlong-distribution/src/main/assemblies/sort-connectors-v1.15.xml @@ -31,7 +31,7 @@ <directory>../inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/jdbc/target</directory> <outputDirectory>inlong-sort/connectors</outputDirectory> <includes> - <include>sort-connector-jdbc-${project.version}-v1.15.jar</include> + <include>sort-connector-postgres-cdc-v1.15-${project.version}.jar</include> </includes> <fileMode>0644</fileMode> </fileSet> diff --git a/inlong-sort/sort-flink/sort-flink-v1.15/pom.xml b/inlong-sort/sort-flink/sort-flink-v1.15/pom.xml index 4d01a9bd0b..6bec4ecaab 100644 --- a/inlong-sort/sort-flink/sort-flink-v1.15/pom.xml +++ b/inlong-sort/sort-flink/sort-flink-v1.15/pom.xml @@ -38,7 +38,7 @@ <properties> <inlong.root.dir>${project.parent.parent.parent.basedir}</inlong.root.dir> <flink.version>1.15.4</flink.version> - <debezium.version>2.0.1.Final</debezium.version> + <debezium.version>1.6.4.Final</debezium.version> <flink.connector.mongodb.cdc.version>2.3.0</flink.connector.mongodb.cdc.version> <flink.connector.oracle.cdc.version>2.3.0</flink.connector.oracle.cdc.version> <flink.connector.mysql.cdc.version>2.3.0</flink.connector.mysql.cdc.version> @@ -53,6 +53,8 @@ <sqlserver.jdbc.version>7.2.2.jre8</sqlserver.jdbc.version> <hudi.version>0.12.3</hudi.version> <thrift.version>0.9.3</thrift.version> + <kafka-clients.version>2.7.1</kafka-clients.version> + <flink-connector-debezium.version>2.3.0</flink-connector-debezium.version> </properties> <dependencyManagement> @@ -95,6 +97,11 @@ <artifactId>flink-connector-sqlserver-cdc</artifactId> <version>${flink.connector.sqlserver.cdc.version}</version> </dependency> + <dependency> + <groupId>com.ververica</groupId> + <artifactId>flink-connector-debezium</artifactId> + <version>${flink-connector-debezium.version}</version> + </dependency> <dependency> <groupId>com.ververica</groupId> <artifactId>flink-connector-mysql-cdc</artifactId> @@ -218,6 +225,11 @@ <artifactId>pulsar-client-all</artifactId> <version>${pulsar.version}</version> </dependency> + <dependency> + <groupId>org.apache.kafka</groupId> + <artifactId>kafka-clients</artifactId> + <version>${kafka-clients.version}</version> + </dependency> </dependencies> </dependencyManagement> diff --git a/inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/pom.xml b/inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/pom.xml index f769edeae2..7b5c1b00a4 100644 --- a/inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/pom.xml +++ b/inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/pom.xml @@ -30,8 +30,20 @@ <packaging>pom</packaging> <name>Apache InLong - Sort Connectors v1.15</name> + <modules> + <module>postgres-cdc</module> + </modules> + <properties> <inlong.root.dir>${project.parent.parent.parent.parent.basedir}</inlong.root.dir> </properties> + <dependencies> + <dependency> + <groupId>org.apache.inlong</groupId> + <artifactId>sort-flink-dependencies-v1.15</artifactId> + <version>${project.version}</version> + </dependency> + </dependencies> + </project> diff --git a/inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/postgres-cdc/pom.xml b/inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/postgres-cdc/pom.xml new file mode 100644 index 0000000000..6e9debb89d --- /dev/null +++ b/inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/postgres-cdc/pom.xml @@ -0,0 +1,150 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + 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. +--> +<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> + <modelVersion>4.0.0</modelVersion> + + <parent> + <groupId>org.apache.inlong</groupId> + <artifactId>sort-connectors-v1.15</artifactId> + <version>1.8.0-SNAPSHOT</version> + </parent> + + <artifactId>sort-connector-postgres-cdc-v1.15</artifactId> + <packaging>jar</packaging> + <name>Apache InLong - Sort-connector-postgres-cdc</name> + + <properties> + <inlong.root.dir>${project.parent.parent.parent.parent.parent.basedir}</inlong.root.dir> + </properties> + + <dependencies> + + <!-- Debezium dependencies --> + <dependency> + <groupId>com.ververica</groupId> + <artifactId>flink-connector-postgres-cdc</artifactId> + </dependency> + + <dependency> + <groupId>com.ververica</groupId> + <artifactId>flink-connector-debezium</artifactId> + <exclusions> + <exclusion> + <groupId>org.apache.kafka</groupId> + <artifactId>kafka-log4j-appender</artifactId> + </exclusion> + </exclusions> + </dependency> + + <dependency> + <groupId>org.apache.kafka</groupId> + <artifactId>kafka-clients</artifactId> + <version>${kafka-clients.version}</version> + </dependency> + + <dependency> + <groupId>io.debezium</groupId> + <artifactId>debezium-connector-postgres</artifactId> + <version>${debezium.version}</version> + <exclusions> + <exclusion> + <groupId>org.postgresql</groupId> + <artifactId>postgresql</artifactId> + </exclusion> + </exclusions> + </dependency> + </dependencies> + + <build> + <plugins> + <plugin> + <groupId>org.apache.maven.plugins</groupId> + <artifactId>maven-shade-plugin</artifactId> + <executions> + <execution> + <id>shade-flink</id> + <goals> + <goal>shade</goal> + </goals> + <phase>package</phase> + <configuration> + <artifactSet> + <includes> + <include>org.apache.inlong:*</include> + <include>io.debezium:debezium-api</include> + <include>io.debezium:debezium-embedded</include> + <include>io.debezium:debezium-core</include> + <include>io.debezium:debezium-connector-postgres</include> + <include>com.ververica:flink-connector-debezium</include> + <include>com.ververica:flink-connector-postgres-cdc</include> + <include>com.google.protobuf:protobuf-java</include> + <include>com.google.guava:*</include> + <include>org.apache.kafka:*</include> + <include>org.postgresql:postgresql</include> + <include>com.fasterxml.*:*</include> + <!-- Include fixed version 18.0-13.0 of flink shaded guava --> + <include>org.apache.flink:flink-shaded-guava</include> + <include>com.google.protobuf:*</include> + </includes> + </artifactSet> + <filters> + <filter> + <artifact>org.apache.inlong:sort-connector-*</artifact> + <includes> + <include>org/apache/inlong/**</include> + <include>META-INF/services/org.apache.flink.table.factories.Factory</include> + </includes> + </filter> + <filter> + <artifact>org.apache.kafka:*</artifact> + <excludes> + <exclude>kafka/kafka-version.properties</exclude> + <exclude>LICENSE</exclude> + <!-- Does not contain anything relevant. + Cites a binary dependency on jersey, but this is neither reflected in the + dependency graph, nor are any jersey files bundled. --> + <exclude>NOTICE</exclude> + <exclude>common/**</exclude> + </excludes> + </filter> + </filters> + <relocations> + <relocation> + <pattern>org.apache.kafka</pattern> + <shadedPattern>com.ververica.cdc.connectors.shaded.org.apache.kafka</shadedPattern> + </relocation> + <relocation> + <pattern>com.google</pattern> + <shadedPattern>com.ververica.cdc.connectors.shaded.com.google</shadedPattern> + </relocation> + <relocation> + <pattern>com.fasterxml</pattern> + <shadedPattern>com.ververica.cdc.connectors.shaded.com.fasterxml</shadedPattern> + </relocation> + </relocations> + </configuration> + </execution> + </executions> + </plugin> + </plugins> + </build> + +</project> diff --git a/inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/postgre/PostgreSQLTableFactory.java b/inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/postgre/PostgreSQLTableFactory.java new file mode 100644 index 0000000000..281add0fae --- /dev/null +++ b/inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/postgre/PostgreSQLTableFactory.java @@ -0,0 +1,181 @@ +/* + * 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.inlong.sort.postgre; + +import com.ververica.cdc.connectors.postgres.table.PostgreSQLTableSource; +import com.ververica.cdc.debezium.table.DebeziumChangelogMode; +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.configuration.ConfigOptions; +import org.apache.flink.configuration.ReadableConfig; +import org.apache.flink.table.catalog.ResolvedSchema; +import org.apache.flink.table.connector.source.DynamicTableSource; +import org.apache.flink.table.factories.DynamicTableFactory; +import org.apache.flink.table.factories.DynamicTableSourceFactory; +import org.apache.flink.table.factories.FactoryUtil; + +import java.util.HashSet; +import java.util.Set; + +import static com.ververica.cdc.debezium.table.DebeziumOptions.DEBEZIUM_OPTIONS_PREFIX; +import static com.ververica.cdc.debezium.table.DebeziumOptions.getDebeziumProperties; +import static com.ververica.cdc.debezium.utils.ResolvedSchemaUtils.getPhysicalSchema; +import static org.apache.flink.util.Preconditions.checkArgument; + +/** Factory for creating configured instance of {@link PostgreSQLTableSource}. */ +public class PostgreSQLTableFactory implements DynamicTableSourceFactory { + + private static final String IDENTIFIER = "postgres-cdc-inlong"; + + private static final ConfigOption<String> HOSTNAME = + ConfigOptions.key("hostname") + .stringType() + .noDefaultValue() + .withDescription("IP address or hostname of the PostgreSQL database server."); + + private static final ConfigOption<Integer> PORT = + ConfigOptions.key("port") + .intType() + .defaultValue(5432) + .withDescription("Integer port number of the PostgreSQL database server."); + + private static final ConfigOption<String> USERNAME = + ConfigOptions.key("username") + .stringType() + .noDefaultValue() + .withDescription( + "Name of the PostgreSQL database to use when connecting to the PostgreSQL database server."); + + private static final ConfigOption<String> PASSWORD = + ConfigOptions.key("password") + .stringType() + .noDefaultValue() + .withDescription( + "Password to use when connecting to the PostgreSQL database server."); + + private static final ConfigOption<String> DATABASE_NAME = + ConfigOptions.key("database-name") + .stringType() + .noDefaultValue() + .withDescription("Database name of the PostgreSQL server to monitor."); + + private static final ConfigOption<String> SCHEMA_NAME = + ConfigOptions.key("schema-name") + .stringType() + .noDefaultValue() + .withDescription("Schema name of the PostgreSQL database to monitor."); + + private static final ConfigOption<String> TABLE_NAME = + ConfigOptions.key("table-name") + .stringType() + .noDefaultValue() + .withDescription("Table name of the PostgreSQL database to monitor."); + + private static final ConfigOption<String> DECODING_PLUGIN_NAME = + ConfigOptions.key("decoding.plugin.name") + .stringType() + .defaultValue("decoderbufs") + .withDescription( + "The name of the Postgres logical decoding plug-in installed on the server.\n" + + "Supported values are decoderbufs, wal2json, wal2json_rds, wal2json_streaming,\n" + + "wal2json_rds_streaming and pgoutput."); + + private static final ConfigOption<String> SLOT_NAME = + ConfigOptions.key("slot.name") + .stringType() + .noDefaultValue() + .withDescription( + "The name of the PostgreSQL logical decoding slot that was created for streaming changes " + + "from a particular plug-in for a particular database/schema. The server uses this slot " + + "to stream events to the connector that you are configuring."); + + private static final ConfigOption<DebeziumChangelogMode> CHANGELOG_MODE = + ConfigOptions.key("changelog-mode") + .enumType(DebeziumChangelogMode.class) + .defaultValue(DebeziumChangelogMode.ALL) + .withDescription( + "The changelog mode used for encoding streaming changes.\n" + + "\"all\": Encodes changes as retract stream using all RowKinds. This is the default mode.\n" + + "\"upsert\": Encodes changes as upsert stream that describes idempotent updates on a key. It can be used for tables with primary keys when replica identity FULL is not an option."); + + @Override + public DynamicTableSource createDynamicTableSource(DynamicTableFactory.Context context) { + final FactoryUtil.TableFactoryHelper helper = + FactoryUtil.createTableFactoryHelper(this, context); + helper.validateExcept(DEBEZIUM_OPTIONS_PREFIX); + + final ReadableConfig config = helper.getOptions(); + String hostname = config.get(HOSTNAME); + String username = config.get(USERNAME); + String password = config.get(PASSWORD); + String databaseName = config.get(DATABASE_NAME); + String schemaName = config.get(SCHEMA_NAME); + String tableName = config.get(TABLE_NAME); + int port = config.get(PORT); + String pluginName = config.get(DECODING_PLUGIN_NAME); + String slotName = config.get(SLOT_NAME); + DebeziumChangelogMode changelogMode = config.get(CHANGELOG_MODE); + ResolvedSchema physicalSchema = + getPhysicalSchema(context.getCatalogTable().getResolvedSchema()); + if (changelogMode == DebeziumChangelogMode.UPSERT) { + checkArgument( + physicalSchema.getPrimaryKey().isPresent(), + "Primary key must be present when upsert mode is selected."); + } + + return new PostgreSQLTableSource( + physicalSchema, + port, + hostname, + databaseName, + schemaName, + tableName, + username, + password, + pluginName, + slotName, + changelogMode, + getDebeziumProperties(context.getCatalogTable().getOptions())); + } + + @Override + public String factoryIdentifier() { + return IDENTIFIER; + } + + @Override + public Set<ConfigOption<?>> requiredOptions() { + Set<ConfigOption<?>> options = new HashSet<>(); + options.add(HOSTNAME); + options.add(USERNAME); + options.add(PASSWORD); + options.add(DATABASE_NAME); + options.add(SCHEMA_NAME); + options.add(TABLE_NAME); + options.add(SLOT_NAME); + return options; + } + + @Override + public Set<ConfigOption<?>> optionalOptions() { + Set<ConfigOption<?>> options = new HashSet<>(); + options.add(PORT); + options.add(DECODING_PLUGIN_NAME); + options.add(CHANGELOG_MODE); + return options; + } +} diff --git a/inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/postgres-cdc/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory b/inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/postgres-cdc/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory new file mode 100644 index 0000000000..06394e72c3 --- /dev/null +++ b/inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/postgres-cdc/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory @@ -0,0 +1,16 @@ +# 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. + +org.apache.inlong.sort.postgre.PostgreSQLTableFactory diff --git a/licenses/inlong-sort-connectors/LICENSE b/licenses/inlong-sort-connectors/LICENSE index 1543e8d655..8ff97a2b93 100644 --- a/licenses/inlong-sort-connectors/LICENSE +++ b/licenses/inlong-sort-connectors/LICENSE @@ -326,400 +326,401 @@ 1.3 Third party Apache 2.0 Licenses ========= -1.3.1 inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/debezium/history/FlinkJsonTableChangeSerializer.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/debezium/internal/DebeziumOffset.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/debezium/internal/DebeziumOffsetSerializer.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/debezium/internal/FlinkDatabaseHistory.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/debezium/internal/FlinkDatabaseSchemaHistory.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/debezium/internal/FlinkOffsetBackingStore.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/debezium/internal/Handover.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/debezium/internal/SchemaRecord.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/debezium/table/AppendMetadataCollector.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/debezium/table/DebeziumOptions.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/debezium/table/DeserializationRuntimeConverter.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/debezium/table/DeserializationRuntimeConverterFactory.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/debezium/table/MetadataConverter.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/debezium/table/RowDataDebeziumDeserializeSchema.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/debezium/DebeziumDeserializationSchema.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/util/DatabaseHistoryUtil.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/util/RecordUtils.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/util/TemporalConversions.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/split/MySqlSnapshotSplit.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/split/MySqlSchemalessSnapshotSplit.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/reader/MySqlSourceReaderContext.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/schema/MySqlTypeUtils.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/utils/StatementUtils.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/schema/MySqlSchema.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/table/MySqlDeserializationConverterFactory.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/connection/JdbcConnectionFactory.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/events/FinishedSnapshotSplitsReportEvent.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/MySqlSource.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/assigners/state/HybridPendingSplitsState.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/connection/ConnectionPoolId.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/debezium/JsonDebeziumDeserializationSchema.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/connection/ConnectionPools.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/split/MySqlRecords.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/debezium/reader/SnapshotSplitReader.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/enumerator/MySqlSourceEnumerator.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/schema/MySqlTableDefinition.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/events/BinlogSplitMetaRequestEvent.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/debezium/Validator.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/debezium/task/context/StatefulTaskContext.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/debezium/DebeziumUtils.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/split/MySqlSplit.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/assigners/MySqlHybridSplitAssigner.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/config/MySqlSourceConfig.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/table/MySqlReadableMetadata.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/table/MySqlTableSource.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/reader/MySqlSourceReader.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/config/MySqlSourceOptions.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/split/MySqlBinlogSplit.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/split/MySqlSnapshotSplitState.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/utils/ChunkUtils.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/events/LatestFinishedSplitsSizeRequestEvent.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/debezium/dispatcher/EventDispatcherImpl.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/debezium/reader/BinlogSplitReader.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/debezium/task/MySqlSnapshotSplitReadTask.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/offset/BinlogOffsetSerializer.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/assigners/ChunkRange.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/utils/TableDiscoveryUtils.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/utils/RecordUtils.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/metrics/MySqlSourceReaderMetrics.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/assigners/state/PendingSplitsStateSerializer.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/assigners/state/BinlogPendingSplitsState.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/debezium/DebeziumSourceFunction.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/reader/MySqlSplitReader.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/debezium/task/context/MySqlErrorHandler.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/assigners/MySqlBinlogSplitAssigner.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/utils/SerializerUtils.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/MySqlSource.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/assigners/AssignerStatus.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/assigners/state/SnapshotPendingSplitsState.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/assigners/MySqlSnapshotSplitAssigner.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/debezium/StringDebeziumDeserializationSchema.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/assigners/ChunkSplitter.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/events/FinishedSnapshotSplitsAckEvent.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/reader/MySqlRecordEmitter.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/offset/BinlogOffset.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/table/StartupMode.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/table/MySqlTableInlongSourceFactory.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/table/OldFieldMetadataConverter.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/utils/ObjectUtils.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/events/BinlogSplitMetaEvent.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/assigners/state/PendingSplitsState.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/config/MySqlSourceConfigFactory.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/split/MySqlSplitState.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/events/SuspendBinlogReaderEvent.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/events/FinishedSnapshotSplitsRequestEvent.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/split/MySqlSplitSerializer.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/table/StartupOptions.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/table/JdbcUrlUtils.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/config/ServerIdRange.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/debezium/EmbeddedFlinkDatabaseHistory.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/split/MySqlBinlogSplitState.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/MySqlValidator.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/events/WakeupReaderEvent.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/debezium/reader/DebeziumReader.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/debezium/dispatcher/SignalEventDispatcher.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/assigners/MySqlSplitAssigner.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/SeekBinlogToTimestampFilter.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/debezium/task/MySqlBinlogSplitReadTask.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/events/LatestFinishedSplitsSizeEvent.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/connection/PooledDataSourceFactory.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/connection/JdbcConnectionPools.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/MySqlSourceBuilder.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/events/SuspendBinlogReaderAckEvent.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/debezium/task/context/MySqlTaskContextImpl.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/schema/MySqlFieldDefinition.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/split/FinishedSnapshotSplitInfo.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/debezium/internal/DebeziumChangeFetcher.java - inlong-sort/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/debezium/internal/DebeziumChangeConsumer.java - inlong-sort/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/cdc/postgres/PostgreSQLSource.java - inlong-sort/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/cdc/postgres/DebeziumSourceFunction.java - inlong-sort/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/cdc/postgres/debezium/internal/ColumnImpl.java - inlong-sort/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/cdc/postgres/debezium/internal/DebeziumChangeFetcher.java - inlong-sort/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/cdc/postgres/debezium/internal/TableEditorImpl.java - inlong-sort/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/cdc/postgres/debezium/internal/TableImpl.java - inlong-sort/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/cdc/postgres/table/PostgreSQLTableFactory.java - inlong-sort/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/cdc/postgres/table/PostgreSQLTableSource.java - inlong-sort/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/cdc/postgres/table/PostgreSQLReadableMetaData.java - inlong-sort/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/MongoDBSource.java - inlong-sort/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/DebeziumSourceFunction.java - inlong-sort/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/table/MongoDBTableSource.java - inlong-sort/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/table/MongoDBTableSourceFactory.java - inlong-sort/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/OracleSource.java - inlong-sort/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/table/OracleTableSource.java - inlong-sort/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/table/OracleTableSourceFactory.java - inlong-sort/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/debezium/internal/DebeziumChangeFetcher.java - inlong-sort/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/debezium/internal/DebeziumChangeConsumer.java - inlong-sort/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/debezium/internal/FlinkDatabaseSchemaHistory.java - inlong-sort/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/debezium/table/RowDataDebeziumDeserializeSchema.java - inlong-sort/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/debezium/DebeziumSourceFunction.java - inlong-sort/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/table/OracleDeserializationConverterFactory.java - inlong-sort/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/table/OracleReadableMetaData.java +1.3.1 inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/debezium/history/FlinkJsonTableChangeSerializer.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/debezium/internal/DebeziumOffset.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/debezium/internal/DebeziumOffsetSerializer.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/debezium/internal/FlinkDatabaseHistory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/debezium/internal/FlinkDatabaseSchemaHistory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/debezium/internal/FlinkOffsetBackingStore.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/debezium/internal/Handover.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/debezium/internal/SchemaRecord.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/debezium/table/AppendMetadataCollector.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/debezium/table/DebeziumOptions.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/debezium/table/DeserializationRuntimeConverter.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/debezium/table/DeserializationRuntimeConverterFactory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/debezium/table/MetadataConverter.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/debezium/table/RowDataDebeziumDeserializeSchema.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/debezium/DebeziumDeserializationSchema.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/util/DatabaseHistoryUtil.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/util/RecordUtils.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/util/TemporalConversions.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/split/MySqlSnapshotSplit.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/split/MySqlSchemalessSnapshotSplit.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/reader/MySqlSourceReaderContext.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/schema/MySqlTypeUtils.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/utils/StatementUtils.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/schema/MySqlSchema.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/table/MySqlDeserializationConverterFactory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/connection/JdbcConnectionFactory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/events/FinishedSnapshotSplitsReportEvent.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/MySqlSource.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/assigners/state/HybridPendingSplitsState.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/connection/ConnectionPoolId.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/debezium/JsonDebeziumDeserializationSchema.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/connection/ConnectionPools.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/split/MySqlRecords.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/debezium/reader/SnapshotSplitReader.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/enumerator/MySqlSourceEnumerator.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/schema/MySqlTableDefinition.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/events/BinlogSplitMetaRequestEvent.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/debezium/Validator.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/debezium/task/context/StatefulTaskContext.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/debezium/DebeziumUtils.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/split/MySqlSplit.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/assigners/MySqlHybridSplitAssigner.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/config/MySqlSourceConfig.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/table/MySqlReadableMetadata.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/table/MySqlTableSource.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/reader/MySqlSourceReader.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/config/MySqlSourceOptions.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/split/MySqlBinlogSplit.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/split/MySqlSnapshotSplitState.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/utils/ChunkUtils.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/events/LatestFinishedSplitsSizeRequestEvent.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/debezium/dispatcher/EventDispatcherImpl.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/debezium/reader/BinlogSplitReader.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/debezium/task/MySqlSnapshotSplitReadTask.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/offset/BinlogOffsetSerializer.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/assigners/ChunkRange.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/utils/TableDiscoveryUtils.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/utils/RecordUtils.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/metrics/MySqlSourceReaderMetrics.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/assigners/state/PendingSplitsStateSerializer.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/assigners/state/BinlogPendingSplitsState.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/debezium/DebeziumSourceFunction.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/reader/MySqlSplitReader.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/debezium/task/context/MySqlErrorHandler.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/assigners/MySqlBinlogSplitAssigner.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/utils/SerializerUtils.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/MySqlSource.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/assigners/AssignerStatus.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/assigners/state/SnapshotPendingSplitsState.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/assigners/MySqlSnapshotSplitAssigner.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/debezium/StringDebeziumDeserializationSchema.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/assigners/ChunkSplitter.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/events/FinishedSnapshotSplitsAckEvent.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/reader/MySqlRecordEmitter.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/offset/BinlogOffset.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/table/StartupMode.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/table/MySqlTableInlongSourceFactory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/table/OldFieldMetadataConverter.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/utils/ObjectUtils.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/events/BinlogSplitMetaEvent.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/assigners/state/PendingSplitsState.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/config/MySqlSourceConfigFactory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/split/MySqlSplitState.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/events/SuspendBinlogReaderEvent.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/events/FinishedSnapshotSplitsRequestEvent.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/split/MySqlSplitSerializer.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/table/StartupOptions.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/table/JdbcUrlUtils.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/config/ServerIdRange.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/debezium/EmbeddedFlinkDatabaseHistory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/split/MySqlBinlogSplitState.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/MySqlValidator.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/events/WakeupReaderEvent.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/debezium/reader/DebeziumReader.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/debezium/dispatcher/SignalEventDispatcher.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/assigners/MySqlSplitAssigner.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/SeekBinlogToTimestampFilter.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/debezium/task/MySqlBinlogSplitReadTask.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/events/LatestFinishedSplitsSizeEvent.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/connection/PooledDataSourceFactory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/connection/JdbcConnectionPools.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/MySqlSourceBuilder.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/events/SuspendBinlogReaderAckEvent.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/debezium/task/context/MySqlTaskContextImpl.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/schema/MySqlFieldDefinition.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/split/FinishedSnapshotSplitInfo.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/debezium/internal/DebeziumChangeFetcher.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/debezium/internal/DebeziumChangeConsumer.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/cdc/postgres/PostgreSQLSource.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/cdc/postgres/DebeziumSourceFunction.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/cdc/postgres/debezium/internal/ColumnImpl.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/cdc/postgres/debezium/internal/DebeziumChangeFetcher.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/cdc/postgres/debezium/internal/TableEditorImpl.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/cdc/postgres/debezium/internal/TableImpl.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/cdc/postgres/table/PostgreSQLTableFactory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/cdc/postgres/table/PostgreSQLTableSource.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/cdc/postgres/table/PostgreSQLReadableMetaData.java + inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/cdc/postgres/table/PostgreSQLTableFactory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/MongoDBSource.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/DebeziumSourceFunction.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/table/MongoDBTableSource.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/table/MongoDBTableSourceFactory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/OracleSource.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/table/OracleTableSource.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/table/OracleTableSourceFactory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/debezium/internal/DebeziumChangeFetcher.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/debezium/internal/DebeziumChangeConsumer.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/debezium/internal/FlinkDatabaseSchemaHistory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/debezium/table/RowDataDebeziumDeserializeSchema.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/debezium/DebeziumSourceFunction.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/table/OracleDeserializationConverterFactory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/table/OracleReadableMetaData.java Source : flink-cdc-connectors 2.2.1 (Please note that the software have been modified.) License : https://github.com/ververica/flink-cdc-connectors/blob/master/LICENSE -1.3.2 inlong-sort/sort-connectors/sort-hive/src/main/java/org/apache/inlong/sort/hive/HiveValidator.java - inlong-sort/sort-connectors/sort-hive/src/main/java/org/apache/inlong/sort/hive/HiveTableSink.java - inlong-sort/sort-connectors/sort-hive/src/main/java/org/apache/inlong/sort/hive/HiveTableMetaStoreFactory.java - inlong-sort/sort-connectors/sort-hive/src/main/java/org/apache/inlong/sort/hive/HiveRowPartitionComputer.java - inlong-sort/sort-connectors/sort-hive/src/main/java/org/apache/inlong/sort/hive/HiveRowDataPartitionComputer.java - inlong-sort/sort-connectors/sort-hive/src/main/java/org/apache/inlong/sort/hive/HiveOptions.java - inlong-sort/sort-connectors/sort-hive/src/main/java/org/apache/inlong/sort/hive/HadoopFileSystemFactory.java - inlong-sort/sort-connectors/sort-hive/src/main/java/org/apache/inlong/sort/hive/table/HiveTableInlongFactory.java - inlong-sort/sort-connectors/sort-hive/src/main/java/org/apache/inlong/sort/hive/filesystem/DefaultHadoopFileCommitterFactory.java - inlong-sort/sort-connectors/sort-hive/src/main/java/org/apache/inlong/sort/hive/filesystem/HadoopPathBasedBulkFormatBuilder.java - inlong-sort/sort-connectors/sort-hive/src/main/java/org/apache/inlong/sort/hive/filesystem/HadoopPathBasedPartFileWriter.java - inlong-sort/sort-connectors/sort-hive/src/main/java/org/apache/inlong/sort/hive/filesystem/HadoopRenameFileCommitter.java - inlong-sort/sort-connectors/sort-hive/src/main/java/org/apache/inlong/sort/hive/filesystem/PartitionCommitter.java - inlong-sort/sort-connectors/sort-hive/src/main/java/org/apache/inlong/sort/hive/HiveBulkWriterFactory.java - inlong-sort/sort-connectors/sort-hive/src/main/java/org/apache/inlong/sort/hive/HiveOutputFormatFactory.java - inlong-sort/sort-connectors/sort-hive/src/main/java/org/apache/inlong/sort/hive/HiveWriterFactory.java +1.3.2 inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-hive/src/main/java/org/apache/inlong/sort/hive/HiveValidator.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-hive/src/main/java/org/apache/inlong/sort/hive/HiveTableSink.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-hive/src/main/java/org/apache/inlong/sort/hive/HiveTableMetaStoreFactory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-hive/src/main/java/org/apache/inlong/sort/hive/HiveRowPartitionComputer.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-hive/src/main/java/org/apache/inlong/sort/hive/HiveRowDataPartitionComputer.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-hive/src/main/java/org/apache/inlong/sort/hive/HiveOptions.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-hive/src/main/java/org/apache/inlong/sort/hive/HadoopFileSystemFactory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-hive/src/main/java/org/apache/inlong/sort/hive/table/HiveTableInlongFactory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-hive/src/main/java/org/apache/inlong/sort/hive/filesystem/DefaultHadoopFileCommitterFactory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-hive/src/main/java/org/apache/inlong/sort/hive/filesystem/HadoopPathBasedBulkFormatBuilder.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-hive/src/main/java/org/apache/inlong/sort/hive/filesystem/HadoopPathBasedPartFileWriter.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-hive/src/main/java/org/apache/inlong/sort/hive/filesystem/HadoopRenameFileCommitter.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-hive/src/main/java/org/apache/inlong/sort/hive/filesystem/PartitionCommitter.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-hive/src/main/java/org/apache/inlong/sort/hive/HiveBulkWriterFactory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-hive/src/main/java/org/apache/inlong/sort/hive/HiveOutputFormatFactory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-hive/src/main/java/org/apache/inlong/sort/hive/HiveWriterFactory.java Source : flink-connector-hive 1.13.5 (Please note that the software have been modified.) License : https://github.com/apache/flink/blob/master/LICENSE -1.3.3 inlong-sort/sort-connectors/sort-pulsar/src/main/java/org/apache/inlong/sort/pulsar/table/DynamicPulsarSerializationSchema.java - inlong-sort/sort-connectors/sort-pulsar/src/main/java/org/apache/inlong/sort/pulsar/table/PulsarDynamicTableFactory.java - inlong-sort/sort-connectors/sort-pulsar/src/main/java/org/apache/inlong/sort/pulsar/table/PulsarDynamicTableSink.java - inlong-sort/sort-connectors/sort-pulsar/src/main/java/org/apache/inlong/sort/pulsar/table/UpsertPulsarDynamicTableFactory.java - inlong-sort/sort-connectors/sort-pulsar/src/main/java/org/apache/inlong/sort/pulsar/internal/FlinkPulsarSource.java - inlong-sort/sort-connectors/sort-pulsar/src/main/java/org/apache/inlong/sort/pulsar/internal/PulsarFetcher.java - inlong-sort/sort-connectors/sort-pulsar/src/main/java/org/apache/inlong/sort/pulsar/internal/PulsarMetadataReader.java - inlong-sort/sort-connectors/sort-pulsar/src/main/java/org/apache/inlong/sort/pulsar/internal/ReaderThread.java - inlong-sort/sort-connectors/sort-pulsar/src/main/java/org/apache/inlong/sort/pulsar/internal/FlinkPulsarSource.java - inlong-sort/sort-connectors/sort-pulsar/src/main/java/org/apache/inlong/sort/pulsar/internal/FlinkPulsarSourceWithoutAdmin.java +1.3.3 inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-pulsar/src/main/java/org/apache/inlong/sort/pulsar/table/DynamicPulsarSerializationSchema.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-pulsar/src/main/java/org/apache/inlong/sort/pulsar/table/PulsarDynamicTableFactory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-pulsar/src/main/java/org/apache/inlong/sort/pulsar/table/PulsarDynamicTableSink.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-pulsar/src/main/java/org/apache/inlong/sort/pulsar/table/UpsertPulsarDynamicTableFactory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-pulsar/src/main/java/org/apache/inlong/sort/pulsar/internal/FlinkPulsarSource.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-pulsar/src/main/java/org/apache/inlong/sort/pulsar/internal/PulsarFetcher.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-pulsar/src/main/java/org/apache/inlong/sort/pulsar/internal/PulsarMetadataReader.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-pulsar/src/main/java/org/apache/inlong/sort/pulsar/internal/ReaderThread.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-pulsar/src/main/java/org/apache/inlong/sort/pulsar/internal/FlinkPulsarSource.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sort-pulsar/src/main/java/org/apache/inlong/sort/pulsar/internal/FlinkPulsarSourceWithoutAdmin.java Source : pulsar-flink-connector_2.11 1.13.6.1-rc9 (Please note that the software have been modified.) License : https://github.com/streamnative/pulsar-flink/blob/master/LICENSE - 1.3.4 inlong-sort/sort-connectors/elasticsearch-6/src/main/java/org/apache/inlong/sort/elasticsearch6/table/Elasticsearch6Configuration.java - inlong-sort/sort-connectors/elasticsearch-6/src/main/java/org/apache/inlong/sort/elasticsearch6/table/Elasticsearch6DynamicSink.java - inlong-sort/sort-connectors/elasticsearch-6/src/main/java/org/apache/inlong/sort/elasticsearch6/table/Elasticsearch6DynamicSinkFactory.java - inlong-sort/sort-connectors/elasticsearch-6/src/main/java/org/apache/inlong/sort/elasticsearch6/utils/RetryRejectedExecutionFailureHandler.java - inlong-sort/sort-connectors/elasticsearch-6/src/test/java/org/apache/inlong/sort/elasticsearch6/table/Elasticsearch6DynamicSinkITCase.java - inlong-sort/sort-connectors/elasticsearch-7/src/main/java/org/apache/inlong/sort/elasticsearch7/table/Elasticsearch7Configuration.java - inlong-sort/sort-connectors/elasticsearch-7/src/main/java/org/apache/inlong/sort/elasticsearch7/table/Elasticsearch7DynamicSink.java - inlong-sort/sort-connectors/elasticsearch-7/src/main/java/org/apache/inlong/sort/elasticsearch7/table/Elasticsearch7DynamicSinkFactory.java - inlong-sort/sort-connectors/elasticsearch-7/src/main/java/org/apache/inlong/sort/elasticsearch7/utils/RetryRejectedExecutionFailureHandler.java - inlong-sort/sort-connectors/elasticsearch-7/src/test/java/org/apache/inlong/sort/elasticsearch7/table/Elasticsearch7DynamicSinkITCase.java - inlong-sort/sort-connectors/elasticsearch-base/src/main/java/org/apache/inlong/sort/elasticsearch/table/AbstractTimeIndexGenerator.java - inlong-sort/sort-connectors/elasticsearch-base/src/main/java/org/apache/inlong/sort/elasticsearch/table/IndexGeneratorFactory.java - inlong-sort/sort-connectors/elasticsearch-base/src/main/java/org/apache/inlong/sort/elasticsearch/table/ElasticsearchConfiguration.java - inlong-sort/sort-connectors/elasticsearch-base/src/main/java/org/apache/inlong/sort/elasticsearch/table/KeyExtractor.java - inlong-sort/sort-connectors/elasticsearch-base/src/main/java/org/apache/inlong/sort/elasticsearch/table/ElasticsearchOptions.java - inlong-sort/sort-connectors/elasticsearch-base/src/main/java/org/apache/inlong/sort/elasticsearch/table/RequestFactory.java - inlong-sort/sort-connectors/elasticsearch-base/src/main/java/org/apache/inlong/sort/elasticsearch/table/ElasticsearchValidationUtils.java - inlong-sort/sort-connectors/elasticsearch-base/src/main/java/org/apache/inlong/sort/elasticsearch/table/IndexGenerator.java - inlong-sort/sort-connectors/elasticsearch-base/src/main/java/org/apache/inlong/sort/elasticsearch/table/RowElasticsearchSinkFunction.java - inlong-sort/sort-connectors/elasticsearch-base/src/main/java/org/apache/inlong/sort/elasticsearch/table/IndexGeneratorBase.java - inlong-sort/sort-connectors/elasticsearch-base/src/main/java/org/apache/inlong/sort/elasticsearch/table/StaticIndexGenerator.java - inlong-sort/sort-connectors/elasticsearch-base/src/main/java/org/apache/inlong/sort/elasticsearch/RequestIndexer.java - inlong-sort/sort-connectors/elasticsearch-base/src/main/java/org/apache/inlong/sort/elasticsearch/ActionRequestFailureHandler.java - inlong-sort/sort-connectors/elasticsearch-base/src/main/java/org/apache/inlong/sort/elasticsearch/utils/NoOpFailureHandler.java - inlong-sort/sort-connectors/elasticsearch-base/src/main/java/org/apache/inlong/sort/elasticsearch/utils/IgnoringFailureHandler.java - inlong-sort/sort-connectors/elasticsearch-base/src/test/java/org/apache/inlong/sort/elasticsearch/table/TestContext.java + 1.3.4 inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/elasticsearch-6/src/main/java/org/apache/inlong/sort/elasticsearch6/table/Elasticsearch6Configuration.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/elasticsearch-6/src/main/java/org/apache/inlong/sort/elasticsearch6/table/Elasticsearch6DynamicSink.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/elasticsearch-6/src/main/java/org/apache/inlong/sort/elasticsearch6/table/Elasticsearch6DynamicSinkFactory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/elasticsearch-6/src/main/java/org/apache/inlong/sort/elasticsearch6/utils/RetryRejectedExecutionFailureHandler.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/elasticsearch-6/src/test/java/org/apache/inlong/sort/elasticsearch6/table/Elasticsearch6DynamicSinkITCase.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/elasticsearch-7/src/main/java/org/apache/inlong/sort/elasticsearch7/table/Elasticsearch7Configuration.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/elasticsearch-7/src/main/java/org/apache/inlong/sort/elasticsearch7/table/Elasticsearch7DynamicSink.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/elasticsearch-7/src/main/java/org/apache/inlong/sort/elasticsearch7/table/Elasticsearch7DynamicSinkFactory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/elasticsearch-7/src/main/java/org/apache/inlong/sort/elasticsearch7/utils/RetryRejectedExecutionFailureHandler.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/elasticsearch-7/src/test/java/org/apache/inlong/sort/elasticsearch7/table/Elasticsearch7DynamicSinkITCase.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/elasticsearch-base/src/main/java/org/apache/inlong/sort/elasticsearch/table/AbstractTimeIndexGenerator.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/elasticsearch-base/src/main/java/org/apache/inlong/sort/elasticsearch/table/IndexGeneratorFactory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/elasticsearch-base/src/main/java/org/apache/inlong/sort/elasticsearch/table/ElasticsearchConfiguration.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/elasticsearch-base/src/main/java/org/apache/inlong/sort/elasticsearch/table/KeyExtractor.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/elasticsearch-base/src/main/java/org/apache/inlong/sort/elasticsearch/table/ElasticsearchOptions.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/elasticsearch-base/src/main/java/org/apache/inlong/sort/elasticsearch/table/RequestFactory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/elasticsearch-base/src/main/java/org/apache/inlong/sort/elasticsearch/table/ElasticsearchValidationUtils.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/elasticsearch-base/src/main/java/org/apache/inlong/sort/elasticsearch/table/IndexGenerator.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/elasticsearch-base/src/main/java/org/apache/inlong/sort/elasticsearch/table/RowElasticsearchSinkFunction.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/elasticsearch-base/src/main/java/org/apache/inlong/sort/elasticsearch/table/IndexGeneratorBase.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/elasticsearch-base/src/main/java/org/apache/inlong/sort/elasticsearch/table/StaticIndexGenerator.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/elasticsearch-base/src/main/java/org/apache/inlong/sort/elasticsearch/RequestIndexer.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/elasticsearch-base/src/main/java/org/apache/inlong/sort/elasticsearch/ActionRequestFailureHandler.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/elasticsearch-base/src/main/java/org/apache/inlong/sort/elasticsearch/utils/NoOpFailureHandler.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/elasticsearch-base/src/main/java/org/apache/inlong/sort/elasticsearch/utils/IgnoringFailureHandler.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/elasticsearch-base/src/test/java/org/apache/inlong/sort/elasticsearch/table/TestContext.java source : flink-connector-elasticsearch 1.13.2 (Please note that the software have been modified.) License : https://github.com/apache/flink/blob/release-1.13.2-rc2/LICENSE - 1.3.5 inlong-sort/sort-connectors/hbase/src/main/java/org/apache/inlong/sort/hbase/HBase2DynamicTableFactory.java - inlong-sort/sort-connectors/hbase/src/main/java/org/apache/inlong/sort/hbase/sink/HBaseSinkFunction.java - inlong-sort/sort-connectors/hbase/src/main/java/org/apache/inlong/sort/hbase/sink/HBaseDynamicTableSink.java + 1.3.5 inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/hbase/src/main/java/org/apache/inlong/sort/hbase/HBase2DynamicTableFactory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/hbase/src/main/java/org/apache/inlong/sort/hbase/sink/HBaseSinkFunction.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/hbase/src/main/java/org/apache/inlong/sort/hbase/sink/HBaseDynamicTableSink.java Source : flink-connector-hbase-2.2 1.13.5 (Please note that the software have been modified.) License : https://github.com/apache/flink/blob/master/LICENSE - 1.3.6 inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/FlinkCatalog.java - inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/FlinkCatalogFactory.java - inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/FlinkConfigOptions.java - inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/FlinkDynamicTableFactory.java - inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/IcebergTableSink.java - inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/BaseDeltaTaskWriter.java - inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/DeltaManifests.java - inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/DeltaManifestsSerializer.java - inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/FlinkManifestUtil.java - inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/FlinkSink.java - inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/EqualityFieldKeySelector.java - inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/IcebergFilesCommitter.java - inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/IcebergStreamWriter.java - inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/ManifestOutputFileFactory.java - inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/PartitionedDeltaWriter.java - inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/PartitionKeySelector.java - inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/RowDataTaskWriterFactory.java - inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/UnpartitionedDeltaWriter.java + 1.3.6 inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/FlinkCatalog.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/FlinkCatalogFactory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/FlinkConfigOptions.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/FlinkDynamicTableFactory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/IcebergTableSink.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/BaseDeltaTaskWriter.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/DeltaManifests.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/DeltaManifestsSerializer.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/FlinkManifestUtil.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/FlinkSink.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/EqualityFieldKeySelector.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/IcebergFilesCommitter.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/IcebergStreamWriter.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/ManifestOutputFileFactory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/PartitionedDeltaWriter.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/PartitionKeySelector.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/RowDataTaskWriterFactory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/UnpartitionedDeltaWriter.java Source : iceberg-flink:iceberg-flink-1.13:0.13.2 (Please note that the software have been modified.) License : https://github.com/apache/iceberg/LICENSE - 1.3.7 inlong-sort/sort-connectors/jdbc/src/main/java/org/apache/inlong/sort/jdbc/table/AbstractJdbcDialect.java - inlong-sort/sort-connectors/jdbc/src/main/java/org/apache/inlong/sort/jdbc/table/JdbcDialects.java - inlong-sort/sort-connectors/jdbc/src/main/java/org/apache/inlong/sort/jdbc/table/JdbcDynamicOutputFormatBuilder.java - inlong-sort/sort-connectors/jdbc/src/main/java/org/apache/inlong/sort/jdbc/table/JdbcDynamicTableFactory.java - inlong-sort/sort-connectors/jdbc/src/main/java/org/apache/inlong/sort/jdbc/table/JdbcDynamicTableSink.java + 1.3.7 inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/jdbc/src/main/java/org/apache/inlong/sort/jdbc/table/AbstractJdbcDialect.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/jdbc/src/main/java/org/apache/inlong/sort/jdbc/table/JdbcDialects.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/jdbc/src/main/java/org/apache/inlong/sort/jdbc/table/JdbcDynamicOutputFormatBuilder.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/jdbc/src/main/java/org/apache/inlong/sort/jdbc/table/JdbcDynamicTableFactory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/jdbc/src/main/java/org/apache/inlong/sort/jdbc/table/JdbcDynamicTableSink.java Source : org.apache.flink:flink-connector-jdbc_2.11:1.13.5 (Please note that the software have been modified.) License : https://github.com/apache/flink/blob/master/LICENSE - 1.3.8 inlong-sort/sort-connectors/sqlserver-cdc/src/main/java/org/apache/inlong/sort/cdc/sqlserver/table/DebeziumSourceFunction.java - inlong-sort/sort-connectors/sqlserver-cdc/src/main/java/org/apache/inlong/sort/cdc/sqlserver/table/SqlServerTableFactory.java - inlong-sort/sort-connectors/sqlserver-cdc/src/main/java/org/apache/inlong/sort/cdc/sqlserver/table/SqlServerTableSource.java - inlong-sort/sort-connectors/sqlserver-cdc/src/main/java/org/apache/inlong/sort/cdc/sqlserver/SqlServerSource.java + 1.3.8 inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sqlserver-cdc/src/main/java/org/apache/inlong/sort/cdc/sqlserver/table/DebeziumSourceFunction.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sqlserver-cdc/src/main/java/org/apache/inlong/sort/cdc/sqlserver/table/SqlServerTableFactory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sqlserver-cdc/src/main/java/org/apache/inlong/sort/cdc/sqlserver/table/SqlServerTableSource.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sqlserver-cdc/src/main/java/org/apache/inlong/sort/cdc/sqlserver/SqlServerSource.java Source : flink-cdc-connectors 2.2.1 (Please note that the software have been modified.) License : https://github.com/ververica/flink-cdc-connectors/blob/master/LICENSE - 1.3.9 inlong-sort/sort-connectors/redis/src/main/java/org/apache/inlong/sort/redis/common/config/handler/FlinkJedisClusterConfigHandler.java - inlong-sort/sort-connectors/redis/src/main/java/org/apache/inlong/sort/redis/common/config/handler/FlinkJedisSentinelConfigHandler.java - inlong-sort/sort-connectors/redis/src/main/java/org/apache/inlong/sort/redis/common/container/RedisCommandsContainerBuilder.java - inlong-sort/sort-connectors/redis/src/main/java/org/apache/inlong/sort/redis/common/handler/RedisMapperHandler.java - inlong-sort/sort-connectors/redis/src/main/java/org/apache/inlong/sort/redis/common/mapper/RedisCommand.java - inlong-sort/sort-connectors/redis/src/main/java/org/apache/inlong/sort/redis/common/mapper/RedisCommandDescription.java - inlong-sort/sort-connectors/redis/src/main/java/org/apache/inlong/sort/redis/common/mapper/RedisMapper.java - inlong-sort/sort-connectors/redis/src/main/java/org/apache/inlong/sort/redis/common/mapper/row/RowRedisMapper.java + 1.3.9 inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/redis/src/main/java/org/apache/inlong/sort/redis/common/config/handler/FlinkJedisClusterConfigHandler.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/redis/src/main/java/org/apache/inlong/sort/redis/common/config/handler/FlinkJedisSentinelConfigHandler.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/redis/src/main/java/org/apache/inlong/sort/redis/common/container/RedisCommandsContainerBuilder.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/redis/src/main/java/org/apache/inlong/sort/redis/common/handler/RedisMapperHandler.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/redis/src/main/java/org/apache/inlong/sort/redis/common/mapper/RedisCommand.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/redis/src/main/java/org/apache/inlong/sort/redis/common/mapper/RedisCommandDescription.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/redis/src/main/java/org/apache/inlong/sort/redis/common/mapper/RedisMapper.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/redis/src/main/java/org/apache/inlong/sort/redis/common/mapper/row/RowRedisMapper.java Source : org.apache.bahir:flink-connector-redis_2.11:1.1-SNAPSHOT (Please note that the software have been modified.) License : https://github.com/apache/bahir-flink/blob/master/LICENSE - 1.3.10 inlong-sort/sort-connectors/filesystem/src/main/java/org/apache/inlong/sort/filesystem/FileSystemTableSink.java - inlong-sort/sort-connectors/filesystem/src/main/java/org/apache/inlong/sort/filesystem/FileSystemTableFactory.java - inlong-sort/sort-connectors/filesystem/src/main/java/org/apache/inlong/sort/filesystem/AbstractFileSystemTable.java - inlong-sort/sort-connectors/filesystem/src/main/java/org/apache/inlong/sort/filesystem/stream/StreamingSink.java - inlong-sort/sort-connectors/filesystem/src/main/java/org/apache/inlong/sort/filesystem/stream/StreamingFileWriter.java - inlong-sort/sort-connectors/filesystem/src/main/java/org/apache/inlong/sort/filesystem/stream/AbstractStreamingWriter.java - inlong-sort/sort-connectors/filesystem/src/main/java/org/apache/inlong/sort/filesystem/stream/compact/CompactFileWriter.java + 1.3.10 inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/filesystem/src/main/java/org/apache/inlong/sort/filesystem/FileSystemTableSink.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/filesystem/src/main/java/org/apache/inlong/sort/filesystem/FileSystemTableFactory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/filesystem/src/main/java/org/apache/inlong/sort/filesystem/AbstractFileSystemTable.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/filesystem/src/main/java/org/apache/inlong/sort/filesystem/stream/StreamingSink.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/filesystem/src/main/java/org/apache/inlong/sort/filesystem/stream/StreamingFileWriter.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/filesystem/src/main/java/org/apache/inlong/sort/filesystem/stream/AbstractStreamingWriter.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/filesystem/src/main/java/org/apache/inlong/sort/filesystem/stream/compact/CompactFileWriter.java Source : flink-table-runtime-blink_2.11-13.2-rc2 2.2.1 (Please note that the software have been modified.) License : https://github.com/apache/flink/blob/release-1.13.2-rc2/LICENSE - 1.3.11 inlong-sort/sort-connectors/kafka/src/main/java/org/apache/inlong/sort/kafka/table/DynamicKafkaDeserializationSchema.java - inlong-sort/sort-connectors/kafka/src/main/java/org/apache/inlong/sort/kafka/table/KafkaDynamicSource.java - inlong-sort/sort-connectors/kafka/src/main/java/org/apache/inlong/sort/kafka/table/KafkaDynamicTableFactory.java - inlong-sort/sort-connectors/kafka/src/main/java/org/apache/inlong/sort/kafka/table/UpsertKafkaDynamicTableFactory.java - inlong-sort/sort-connectors/kafka/src/main/java/org/apache/inlong/sort/kafka/DynamicKafkaSerializationSchema.java - inlong-sort/sort-connectors/kafka/src/main/java/org/apache/inlong/sort/kafka/FlinkKafkaConsumer.java - inlong-sort/sort-connectors/kafka/src/main/java/org/apache/inlong/sort/kafka/FlinkKafkaConsumerBase.java - inlong-sort/sort-connectors/kafka/src/main/java/org/apache/inlong/sort/kafka/FlinkKafkaProducer.java - inlong-sort/sort-connectors/kafka/src/main/java/org/apache/inlong/sort/kafka/KafkaDynamicSink.java + 1.3.11 inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/kafka/src/main/java/org/apache/inlong/sort/kafka/table/DynamicKafkaDeserializationSchema.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/kafka/src/main/java/org/apache/inlong/sort/kafka/table/KafkaDynamicSource.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/kafka/src/main/java/org/apache/inlong/sort/kafka/table/KafkaDynamicTableFactory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/kafka/src/main/java/org/apache/inlong/sort/kafka/table/UpsertKafkaDynamicTableFactory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/kafka/src/main/java/org/apache/inlong/sort/kafka/DynamicKafkaSerializationSchema.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/kafka/src/main/java/org/apache/inlong/sort/kafka/FlinkKafkaConsumer.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/kafka/src/main/java/org/apache/inlong/sort/kafka/FlinkKafkaConsumerBase.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/kafka/src/main/java/org/apache/inlong/sort/kafka/FlinkKafkaProducer.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/kafka/src/main/java/org/apache/inlong/sort/kafka/KafkaDynamicSink.java Source : org.apache.flink:flink-connector-kafka_2.11:1.13.5 (Please note that the software have been modified.) License : https://github.com/apache/flink/blob/master/LICENSE - 1.3.12 inlong-sort/sort-connectors/doris/src/main/java/org/apache/inlong/sort/doris/table/DorisDynamicSchemaOutputFormat.java - inlong-sort/sort-connectors/doris/src/main/java/org/apache/inlong/sort/doris/table/DorisDynamicTableFactory.java - inlong-sort/sort-connectors/doris/src/main/java/org/apache/inlong/sort/doris/table/DorisDynamicTableSink.java - inlong-sort/sort-connectors/doris/src/main/java/org/apache/inlong/sort/doris/table/DorisStreamLoad.java - inlong-sort/sort-connectors/doris/src/main/java/org/apache/inlong/sort/doris/model/RespContent.java + 1.3.12 inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/doris/src/main/java/org/apache/inlong/sort/doris/table/DorisDynamicSchemaOutputFormat.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/doris/src/main/java/org/apache/inlong/sort/doris/table/DorisDynamicTableFactory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/doris/src/main/java/org/apache/inlong/sort/doris/table/DorisDynamicTableSink.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/doris/src/main/java/org/apache/inlong/sort/doris/table/DorisStreamLoad.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/doris/src/main/java/org/apache/inlong/sort/doris/model/RespContent.java Source : org.apache.doris:flink-doris-connector-1.13_2.11:1.0.3 (Please note that the software have been modified.) License : https://github.com/apache/doris-flink-connector/blob/1.13_2.11-1.0.3/LICENSE.txt -1.3.13 inlong-sort/sort-connectors/starrocks/src/main/java/org/apache/inlong/sort/starrocks/manager/StarRocksSinkManager.java - inlong-sort/sort-connectors/starrocks/src/main/java/org/apache/inlong/sort/starrocks/manager/StarRocksStreamLoadVisitor.java - inlong-sort/sort-connectors/starrocks/src/main/java/org/apache/inlong/sort/starrocks/table/sink/StarRocksDynamicSinkFunction.java - inlong-sort/sort-connectors/starrocks/src/main/java/org/apache/inlong/sort/starrocks/table/sink/StarRocksDynamicTableSink.java - inlong-sort/sort-connectors/starrocks/src/main/java/org/apache/inlong/sort/starrocks/table/sink/StarRocksDynamicTableSinkFactory.java +1.3.13 inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/starrocks/src/main/java/org/apache/inlong/sort/starrocks/manager/StarRocksSinkManager.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/starrocks/src/main/java/org/apache/inlong/sort/starrocks/manager/StarRocksStreamLoadVisitor.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/starrocks/src/main/java/org/apache/inlong/sort/starrocks/table/sink/StarRocksDynamicSinkFunction.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/starrocks/src/main/java/org/apache/inlong/sort/starrocks/table/sink/StarRocksDynamicTableSink.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/starrocks/src/main/java/org/apache/inlong/sort/starrocks/table/sink/StarRocksDynamicTableSinkFactory.java Source : com.starrocks:flink-connector-starrocks:1.2.3_flink-1.13_2.11 (Please note that the software have been modified.) License : https://www.apache.org/licenses/LICENSE-2.0.txt -1.3.14 inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/jdbc/JdbcIncrementalSource.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/reader/IncrementalSourceRecordEmitter.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/IncrementalSource.java - inlong-sort/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/source/OracleSourceBuilder.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/config/BaseSourceConfig.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/config/JdbcSourceConfig.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/config/JdbcSourceConfigFactory.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/config/SourceConfig.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/dialect/DataSourceDialect.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/dialect/JdbcDataSourceDialect.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/relational/connection/ConnectionPoolId.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/relational/connection/ConnectionPools.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/relational/connection/JdbcConnectionFactory.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/relational/connection/JdbcConnectionPoolFactory.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/relational/connection/JdbcConnectionPools.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/relational/JdbcSourceEventDispatcher.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/assigner/splitter/ChunkRange.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/assigner/splitter/ChunkSplitter.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/assigner/splitter/JdbcSourceChunkSplitter.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/assigner/state/HybridPendingSplitsState.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/assigner/state/PendingSplitsState.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/assigner/state/PendingSplitsStateSerializer.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/assigner/state/SnapshotPendingSplitsState.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/assigner/state/StreamPendingSplitsState.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/assigner/HybridSplitAssigner.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/assigner/SnapshotSplitAssigner.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/assigner/SplitAssigner.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/assigner/StreamSplitAssigner.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/enumerator/IncrementalSourceEnumerator.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/jdbc/JdbcIncrementalSource.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/events/FinishedSnapshotSplitsAckEvent.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/events/FinishedSnapshotSplitsReportEvent.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/events/FinishedSnapshotSplitsRequestEvent.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/events/StreamSplitMetaEvent.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/events/StreamSplitMetaRequestEvent.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/offset/Offset.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/offset/OffsetDeserializerSerializer.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/offset/OffsetFactory.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/split/ChangeEventRecords.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/split/FinishedSnapshotSplitInfo.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/split/MetricSplit.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/split/SchemalessSnapshotSplit.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/split/SnapshotSplit.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/split/SnapshotSplitState.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/split/SourceRecords.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/split/SourceSplitBase.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/split/SourceSplitSerializer.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/split/SourceSplitState.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/split/StreamSplit.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/split/StreamSplitState.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/wartermark/WatermarkEvent.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/wartermark/WatermarkKind.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/metrics/SourceReaderMetrics.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/reader/external/Fetcher.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/reader/external/FetchTask.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/reader/external/IncrementalSourceScanFetcher.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/reader/external/IncrementalSourceStreamFetcher.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/reader/external/JdbcSourceFetchTaskContext.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/reader/IncrementalSourceReader.java - inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/reader/IncrementalSourceSplitReader.java - inlong-sort/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/base/relational/JdbcSourceEventDispatcher.java - inlong-sort/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/source/config/OracleSourceConfig.java - inlong-sort/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/source/config/OracleSourceConfigFactory.java - inlong-sort/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/source/config/OracleSourceOptions.java - inlong-sort/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/source/meta/offset/RedoLogOffset.java - inlong-sort/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/source/meta/offset/RedoLogOffsetFactory.java - inlong-sort/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/source/reader/fetch/OracleScanFetchTask.java - inlong-sort/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/source/reader/fetch/OracleSourceFetchTaskContext.java - inlong-sort/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/source/reader/fetch/OracleStreamFetchTask.java - inlong-sort/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/source/splitter/OracleChunkSplitter.java - inlong-sort/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/source/utils/OracleConnectionUtils.java - inlong-sort/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/source/utils/OracleSchema.java - inlong-sort/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/source/utils/OracleTypeUtils.java - inlong-sort/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/source/utils/OracleUtils.java - inlong-sort/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/source/OracleDialect.java - inlong-sort/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/source/OraclePooledDataSourceFactory.java - inlong-sort/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/base/source/IncrementalSource.java - inlong-sort/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/assigners/splitters/ShardedSplitStrategy.java - inlong-sort/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/MongoDBSource.java - inlong-sort/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/assigners/splitters/SplitVectorSplitStrategy.java - inlong-sort/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/assigners/splitters/SingleSplitStrategy.java - inlong-sort/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/assigners/splitters/MongoDBChunkSplitter.java - inlong-sort/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/assigners/splitters/SplitContext.java - inlong-sort/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/assigners/splitters/SampleBucketSplitStrategy.java - inlong-sort/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/base/source/IncrementalSource.java - inlong-sort/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/assigners/splitters/SplitStrategy.java - inlong-sort/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/MongoDBSourceBuilder.java - inlong-sort/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/config/MongoDBSourceConfig.java - inlong-sort/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/config/MongoDBSourceConfigFactory.java - inlong-sort/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/config/MongoDBSourceOptions.java - inlong-sort/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/connection/MongoClientPool.java - inlong-sort/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/dialect/MongoDBDialect.java - inlong-sort/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/offset/ChangeStreamDescriptor.java - inlong-sort/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/offset/ChangeStreamOffset.java - inlong-sort/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/offset/ChangeStreamOffsetFactory.java - inlong-sort/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/reader/fetch/MongoDBFetchTaskContext.java - inlong-sort/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/reader/fetch/MongoDBScanFetchTask.java - inlong-sort/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/reader/fetch/MongoDBStreamFetchTask.java - inlong-sort/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/reader/MongoDBRecordEmitter.java - inlong-sort/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/utils/MongoUtils.java +1.3.14 inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/jdbc/JdbcIncrementalSource.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/reader/IncrementalSourceRecordEmitter.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/IncrementalSource.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/source/OracleSourceBuilder.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/config/BaseSourceConfig.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/config/JdbcSourceConfig.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/config/JdbcSourceConfigFactory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/config/SourceConfig.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/dialect/DataSourceDialect.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/dialect/JdbcDataSourceDialect.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/relational/connection/ConnectionPoolId.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/relational/connection/ConnectionPools.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/relational/connection/JdbcConnectionFactory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/relational/connection/JdbcConnectionPoolFactory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/relational/connection/JdbcConnectionPools.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/relational/JdbcSourceEventDispatcher.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/assigner/splitter/ChunkRange.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/assigner/splitter/ChunkSplitter.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/assigner/splitter/JdbcSourceChunkSplitter.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/assigner/state/HybridPendingSplitsState.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/assigner/state/PendingSplitsState.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/assigner/state/PendingSplitsStateSerializer.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/assigner/state/SnapshotPendingSplitsState.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/assigner/state/StreamPendingSplitsState.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/assigner/HybridSplitAssigner.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/assigner/SnapshotSplitAssigner.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/assigner/SplitAssigner.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/assigner/StreamSplitAssigner.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/enumerator/IncrementalSourceEnumerator.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/jdbc/JdbcIncrementalSource.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/events/FinishedSnapshotSplitsAckEvent.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/events/FinishedSnapshotSplitsReportEvent.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/events/FinishedSnapshotSplitsRequestEvent.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/events/StreamSplitMetaEvent.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/events/StreamSplitMetaRequestEvent.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/offset/Offset.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/offset/OffsetDeserializerSerializer.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/offset/OffsetFactory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/split/ChangeEventRecords.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/split/FinishedSnapshotSplitInfo.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/split/MetricSplit.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/split/SchemalessSnapshotSplit.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/split/SnapshotSplit.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/split/SnapshotSplitState.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/split/SourceRecords.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/split/SourceSplitBase.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/split/SourceSplitSerializer.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/split/SourceSplitState.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/split/StreamSplit.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/split/StreamSplitState.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/wartermark/WatermarkEvent.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/wartermark/WatermarkKind.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/metrics/SourceReaderMetrics.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/reader/external/Fetcher.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/reader/external/FetchTask.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/reader/external/IncrementalSourceScanFetcher.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/reader/external/IncrementalSourceStreamFetcher.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/reader/external/JdbcSourceFetchTaskContext.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/reader/IncrementalSourceReader.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/src/main/java/org/apache/inlong/sort/cdc/base/source/reader/IncrementalSourceSplitReader.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/base/relational/JdbcSourceEventDispatcher.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/source/config/OracleSourceConfig.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/source/config/OracleSourceConfigFactory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/source/config/OracleSourceOptions.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/source/meta/offset/RedoLogOffset.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/source/meta/offset/RedoLogOffsetFactory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/source/reader/fetch/OracleScanFetchTask.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/source/reader/fetch/OracleSourceFetchTaskContext.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/source/reader/fetch/OracleStreamFetchTask.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/source/splitter/OracleChunkSplitter.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/source/utils/OracleConnectionUtils.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/source/utils/OracleSchema.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/source/utils/OracleTypeUtils.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/source/utils/OracleUtils.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/source/OracleDialect.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/source/OraclePooledDataSourceFactory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/base/source/IncrementalSource.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/assigners/splitters/ShardedSplitStrategy.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/MongoDBSource.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/assigners/splitters/SplitVectorSplitStrategy.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/assigners/splitters/SingleSplitStrategy.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/assigners/splitters/MongoDBChunkSplitter.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/assigners/splitters/SplitContext.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/assigners/splitters/SampleBucketSplitStrategy.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/base/source/IncrementalSource.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/assigners/splitters/SplitStrategy.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/MongoDBSourceBuilder.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/config/MongoDBSourceConfig.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/config/MongoDBSourceConfigFactory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/config/MongoDBSourceOptions.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/connection/MongoClientPool.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/dialect/MongoDBDialect.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/offset/ChangeStreamDescriptor.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/offset/ChangeStreamOffset.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/offset/ChangeStreamOffsetFactory.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/reader/fetch/MongoDBFetchTaskContext.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/reader/fetch/MongoDBScanFetchTask.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/reader/fetch/MongoDBStreamFetchTask.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/reader/MongoDBRecordEmitter.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/utils/MongoUtils.java Source : flink-cdc-connectors 2.3.0 (Please note that the software have been modified.) License : https://github.com/ververica/flink-cdc-connectors/blob/master/LICENSE - 1.3.15 inlong-sort/sort-connectors/oracle-cdc/src/main/java/io/debezium/connector/oracle/logminer/LogMinerQueryBuilder.java - inlong-sort/sort-connectors/oracle-cdc/src/main/java/io/debezium/connector/oracle/logminer/LogMinerStreamingChangeEventSource.java + 1.3.15 inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/io/debezium/connector/oracle/logminer/LogMinerQueryBuilder.java + inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/io/debezium/connector/oracle/logminer/LogMinerStreamingChangeEventSource.java Source : debezium-connector-oracle 1.6.4.Final (Please note that the software have been modified.) License : https://github.com/debezium/debezium/blob/main/LICENSE.txt