[Feature][Connector-V2] Add tail file source

Author: hailin0

.github/workflows/labeler/label-scope-conf.yml

@@ -310,4 +310,10 @@ aerospike:
   - all:
       - changed-files:
           - any-glob-to-any-file: seatunnel-connectors-v2/connector-aerospike/**
-          - all-globs-to-all-files: '!seatunnel-connectors-v2/connector-!(aerospike)/**'
+          - all-globs-to-all-files: '!seatunnel-connectors-v2/connector-!(aerospike)/**'
+
+tailfile:
+  - all:
+      - changed-files:
+          - any-glob-to-any-file: seatunnel-connectors-v2/connector-tailfile/**
+          - all-globs-to-all-files: '!seatunnel-connectors-v2/connector-!(tailfile)/**'

config/plugin_config

@@ -93,3 +93,4 @@ connector-sls
 connector-qdrant
 connector-typesense
 connector-cdc-opengauss
+connector-tailfile

plugin-mapping.properties

@@ -142,7 +142,7 @@ seatunnel.source.Typesense = connector-typesense
 seatunnel.sink.Typesense = connector-typesense
 seatunnel.source.Opengauss-CDC = connector-cdc-opengauss
 seatunnel.sink.Aerospike = connector-aerospike
-
+seatunnel.source.tailfile = connector-tailfile
 seatunnel.transform.Sql = seatunnel-transforms-v2
 seatunnel.transform.FieldMapper = seatunnel-transforms-v2
 seatunnel.transform.Filter = seatunnel-transforms-v2

seatunnel-api/src/main/java/org/apache/seatunnel/api/source/SourceSplitEnumerator.java

@@ -41,6 +41,11 @@ public interface SourceSplitEnumerator<SplitT extends SourceSplit, StateT>
     /** The method is executed by the engine only once. */
     void run() throws Exception;
 
+    default boolean runStep() throws Exception {
+        run();
+        return false;
+    }
+
     /**
      * Called to close the enumerator, in case it holds on to any resources, like threads or network
      * connections.

seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/reader/external/JdbcSourceFetchTaskContext.java

@@ -27,6 +27,7 @@
 import org.apache.seatunnel.connectors.cdc.base.source.split.SourceSplitBase;
 import org.apache.seatunnel.connectors.cdc.base.utils.SourceRecordUtils;
 import org.apache.seatunnel.connectors.cdc.debezium.ConnectTableChangeSerializer;
+import org.apache.seatunnel.connectors.cdc.debezium.DebeziumSchemaNameAdjuster;
 import org.apache.seatunnel.connectors.cdc.debezium.EmbeddedDatabaseHistory;
 
 import org.apache.kafka.connect.data.SchemaAndValue;
@@ -71,7 +72,7 @@ public JdbcSourceFetchTaskContext(
         this.sourceConfig = sourceConfig;
         this.dataSourceDialect = dataSourceDialect;
         this.dbzConnectorConfig = sourceConfig.getDbzConnectorConfig();
-        this.schemaNameAdjuster = SchemaNameAdjuster.create();
+        this.schemaNameAdjuster = DebeziumSchemaNameAdjuster.create();
         this.jsonConverter = new JsonConverter();
         jsonConverter.configure(Collections.singletonMap("schemas.enable", true), false);
     }

seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/debezium/ConnectTableChangeSerializer.java

@@ -63,7 +63,8 @@
 public class ConnectTableChangeSerializer
         implements TableChanges.TableChangesSerializer<List<Struct>>, Serializable {
     private static final String ENUM_VALUES_KEY = "enumValues";
-    private static final SchemaNameAdjuster SCHEMA_NAME_ADJUSTER = SchemaNameAdjuster.create();
+    private static final SchemaNameAdjuster SCHEMA_NAME_ADJUSTER =
+            DebeziumSchemaNameAdjuster.create();
 
     private static final Schema COLUMN_SCHEMA =
             SchemaBuilder.struct()

seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/debezium/DebeziumSchemaNameAdjuster.java

@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.cdc.debezium;
+
+import org.apache.kafka.connect.errors.ConnectException;
+
+import io.debezium.util.SchemaNameAdjuster;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class DebeziumSchemaNameAdjuster {
+    private static final SchemaNameAdjuster.ReplacementOccurred HANDLER =
+            (original, replacement, conflictsWith) -> {
+                if (conflictsWith != null) {
+                    String msg =
+                            "The Kafka Connect schema name '"
+                                    + original
+                                    + "' is not a valid Avro schema name and its replacement '"
+                                    + replacement
+                                    + "' conflicts with another different schema '"
+                                    + conflictsWith
+                                    + "'";
+                    log.error(msg);
+                    throw new ConnectException(msg);
+                } else {
+                    log.warn(
+                            "The Kafka Connect schema name '{}' is not a valid Avro schema name, so replacing with '{}'",
+                            original,
+                            replacement);
+                }
+            };
+
+    private static final SchemaNameAdjuster.ReplacementFunction REPLACE_CHAR_HANDLER =
+            invalid -> {
+                // Support for Chinese characters
+                if (Character.isIdeographic(invalid)) {
+                    return String.valueOf(invalid);
+                }
+                return "_";
+            };
+
+    public static SchemaNameAdjuster create() {
+        return (original) ->
+                SchemaNameAdjuster.validFullname(
+                        original, REPLACE_CHAR_HANDLER, HANDLER.firstTimeOnly());
+    }
+}

seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/io/debezium/connector/oracle/logminer/LogMinerStreamingChangeEventSource.java

@@ -122,7 +122,10 @@ public LogMinerStreamingChangeEventSource(
         this.schema = schema;
         this.connectorConfig = connectorConfig;
         this.strategy = connectorConfig.getLogMiningStrategy();
-        this.isContinuousMining = connectorConfig.isContinuousMining();
+        this.isContinuousMining =
+                jdbcConnection.getOracleVersion().getMajor() >= 19
+                        ? false
+                        : connectorConfig.isContinuousMining();
         this.errorHandler = errorHandler;
         this.streamingMetrics = streamingMetrics;
         this.jdbcConfiguration = JdbcConfiguration.adapt(jdbcConfig);

seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/io/debezium/connector/postgresql/PostgresObjectUtils.java

@@ -17,8 +17,6 @@
 
 package io.debezium.connector.postgresql;
 
-import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException;
-
 import org.apache.kafka.connect.errors.ConnectException;
 
 import io.debezium.connector.postgresql.connection.PostgresConnection;
@@ -32,8 +30,6 @@
 import java.sql.SQLException;
 import java.time.Duration;
 
-import static org.apache.seatunnel.connectors.seatunnel.cdc.postgres.exception.PostgresConnectorErrorCode.CREATE_REPLICATION_CONNECTION_FAILED;
-
 /**
  * A factory for creating various Debezium objects
  *
@@ -118,6 +114,6 @@ public static ReplicationConnection createReplicationConnection(
                 }
             }
         }
-        throw new SeaTunnelRuntimeException(CREATE_REPLICATION_CONNECTION_FAILED, "" + taskContext);
+        throw new RuntimeException("Failed to create replication connection" + taskContext);
     }
 }

seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/io/debezium/connector/postgresql/PostgresOffsetContext.java

@@ -253,8 +253,10 @@ public PostgresOffsetContext load(Map<String, ?> offset) {
             final Long txId = readOptionalLong(offset, SourceInfo.TXID_KEY);
 
             final Instant useconds =
-                    Conversions.toInstantFromMicros(
-                            (Long) offset.get(SourceInfo.TIMESTAMP_USEC_KEY));
+                    offset.get(SourceInfo.TIMESTAMP_USEC_KEY) != null
+                            ? Conversions.toInstantFromMicros(
+                                    (Long) offset.get(SourceInfo.TIMESTAMP_USEC_KEY))
+                            : Clock.system().currentTimeAsInstant();
             final boolean snapshot =
                     (boolean)
                             ((Map<String, Object>) offset)