Merge branch 'apache:master' into master

Author: beryllw

.github/labeler.yml

@@ -84,3 +84,5 @@ doris-pipeline-connector:
   - flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-doris/**/*
 starrocks-pipeline-connector:
   - flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/**/*
+elasticsearch-pipeline-connector:
+  - flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-elasticsearch/**/*

.github/workflows/flink_cdc.yml

@@ -94,8 +94,10 @@ env:
   flink-cdc-connect/flink-cdc-source-connectors/flink-connector-vitess-cdc,\
   flink-cdc-connect/flink-cdc-source-connectors/flink-sql-connector-vitess-cdc"
 
-  MODULES_E2E: "\
-  flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests,\
+  MODULES_PIPELINE_E2E: "\
+  flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests"
+
+  MODULES_SOURCE_E2E: "\
   flink-cdc-e2e-tests/flink-cdc-source-e2e-tests"
 
 jobs:
@@ -134,7 +136,8 @@ jobs:
                   "oceanbase",
                   "db2",
                   "vitess",
-                  "e2e"
+                  "pipeline_e2e",
+                  "source_e2e"
         ]
     timeout-minutes: 120
     env:
@@ -149,6 +152,23 @@ jobs:
     steps:
       - run: echo "Running CI pipeline for JDK version ${{ matrix.jdk }}"
 
+      - name: Clean up disk space
+        run: |
+          set -euo pipefail
+          
+          echo "Disk space before cleanup"
+          df -h 
+          
+          echo "Cleaning up disk space"
+          sudo rm -rf /usr/share/dotnet
+          sudo rm -rf /usr/local/lib/android
+          sudo rm -rf /opt/ghc
+          sudo rm -rf /opt/hostedtoolcache/CodeQL
+          sudo docker image prune --all --force
+          
+          echo "Disk space after cleanup"
+          df -h
+
       - name: Check out repository code
         uses: actions/checkout@v4
         with:
@@ -205,13 +225,17 @@ jobs:
               ("vitess")
                modules=${{ env.MODULES_VITESS }}
               ;;
-              ("e2e")
-               compile_modules="${{ env.MODULES_CORE }},${{ env.MODULES_PIPELINE_CONNECTORS }},${{ env.MODULES_MYSQL }},${{ env.MODULES_POSTGRES }},${{ env.MODULES_ORACLE }},${{ env.MODULES_MONGODB }},${{ env.MODULES_SQLSERVER }},${{ env.MODULES_TIDB }},${{ env.MODULES_OCEANBASE }},${{ env.MODULES_DB2 }},${{ env.MODULES_VITESS }},${{ env.MODULES_E2E }}"
-               modules=${{ env.MODULES_E2E }}
+              ("pipeline_e2e")
+               compile_modules="${{ env.MODULES_CORE }},${{ env.MODULES_PIPELINE_CONNECTORS }},${{ env.MODULES_MYSQL }},${{ env.MODULES_POSTGRES }},${{ env.MODULES_ORACLE }},${{ env.MODULES_MONGODB }},${{ env.MODULES_SQLSERVER }},${{ env.MODULES_TIDB }},${{ env.MODULES_OCEANBASE }},${{ env.MODULES_DB2 }},${{ env.MODULES_VITESS }},${{ env.MODULES_PIPELINE_E2E }}"
+               modules=${{ env.MODULES_PIPELINE_E2E }}
+              ;;
+              ("source_e2e")
+               compile_modules="${{ env.MODULES_CORE }},${{ env.MODULES_PIPELINE_CONNECTORS }},${{ env.MODULES_MYSQL }},${{ env.MODULES_POSTGRES }},${{ env.MODULES_ORACLE }},${{ env.MODULES_MONGODB }},${{ env.MODULES_SQLSERVER }},${{ env.MODULES_TIDB }},${{ env.MODULES_OCEANBASE }},${{ env.MODULES_DB2 }},${{ env.MODULES_VITESS }},${{ env.MODULES_SOURCE_E2E }}"
+               modules=${{ env.MODULES_SOURCE_E2E }}
               ;;
           esac
 
-          if [ ${{ matrix.module }} != "e2e" ]; then
+          if [ ${{ matrix.module }} != "pipeline_e2e" ] && [ ${{ matrix.module }} != "source_e2e" ]; then
             compile_modules=$modules
           fi
 

Dockerfile

@@ -0,0 +1,35 @@
+#/*
+# * Licensed to the Apache Software Foundation (ASF) under one or more
+# * contributor license agreements.  See the NOTICE file distributed with
+# * this work for additional information regarding copyright ownership.
+# * The ASF licenses this file to You under the Apache License, Version 2.0
+# * (the "License"); you may not use this file except in compliance with
+# * the License.  You may obtain a copy of the License at
+# *
+# *      http://www.apache.org/licenses/LICENSE-2.0
+# *
+# * Unless required by applicable law or agreed to in writing, software
+# * distributed under the License is distributed on an "AS IS" BASIS,
+# * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# * See the License for the specific language governing permissions and
+# * limitations under the License.
+# */
+
+FROM flink
+
+ARG FLINK_CDC_VERSION=3.3-SNAPSHOT
+ARG PIPELINE_DEFINITION_FILE
+
+RUN mkdir -p /opt/flink-cdc
+RUN mkdir -p /opt/flink/usrlib
+ENV FLINK_CDC_HOME /opt/flink-cdc
+COPY flink-cdc-dist/target/flink-cdc-${FLINK_CDC_VERSION}-bin.tar.gz /tmp/
+RUN tar -xzvf /tmp/flink-cdc-${FLINK_CDC_VERSION}-bin.tar.gz -C /tmp/ && \
+    mv /tmp/flink-cdc-${FLINK_CDC_VERSION}/* /opt/flink-cdc/ && \
+    mv /opt/flink-cdc/lib/flink-cdc-dist-${FLINK_CDC_VERSION}.jar /opt/flink-cdc/lib/flink-cdc-dist.jar && \
+    rm -rf /tmp/flink-cdc-${FLINK_CDC_VERSION} /tmp/flink-cdc-${FLINK_CDC_VERSION}-bin.tar.gz
+# copy jars to cdc libs
+COPY flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-values/target/flink-cdc-pipeline-connector-values-${FLINK_CDC_VERSION}.jar /opt/flink/usrlib/flink-cdc-pipeline-connector-values-${FLINK_CDC_VERSION}.jar
+COPY flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/target/flink-cdc-pipeline-connector-mysql-${FLINK_CDC_VERSION}.jar /opt/flink/usrlib/flink-cdc-pipeline-connector-mysql-${FLINK_CDC_VERSION}.jar
+# copy flink cdc pipeline conf file, Here is an example. Users can replace it according to their needs.
+COPY $PIPELINE_DEFINITION_FILE $FLINK_CDC_HOME/conf

README.md

@@ -35,41 +35,46 @@ full database synchronization, sharding table synchronization, schema evolution
 2. [Download](https://github.com/apache/flink-cdc/releases) Flink CDC tar, unzip it and put jars of pipeline connector to Flink `lib` directory.
 3. Create a **YAML** file to describe the data source and data sink, the following example synchronizes all tables under MySQL app_db database to Doris :
   ```yaml
-    source:
-       type: mysql
-       name: MySQL Source
-       hostname: 127.0.0.1
-       port: 3306
-       username: admin
-       password: pass
-       tables: adb.\.*
-       server-id: 5401-5404
-    
-    sink:
-      type: doris
-      name: Doris Sink
-      fenodes: 127.0.0.1:8030
-      username: root
-      password: pass
-      
-    transform:
-      - source-table: adb.web_order01
-        projection: \*, UPPER(product_name) as product_name
-        filter: id > 10 AND order_id > 100
-        description: project fields and filter
-      - source-table: adb.web_order02
-        projection: \*, UPPER(product_name) as product_name
-        filter: id > 20 AND order_id > 200
-        description: project fields and filter  
-
-    route:
-      - source-table: adb.web_order\.*
-        sink-table: adb.ods_web_orders
-        description: sync sharding tables to one destination table
-    
-    pipeline:
-       name: MySQL to Doris Pipeline
-       parallelism: 4
+   source:
+     type: mysql
+     hostname: localhost
+     port: 3306
+     username: root
+     password: 123456
+     tables: app_db.\.*
+
+   sink:
+     type: doris
+     fenodes: 127.0.0.1:8030
+     username: root
+     password: ""
+
+   transform:
+     - source-table: adb.web_order01
+       projection: \*, format('%S', product_name) as product_name
+       filter: addone(id) > 10 AND order_id > 100
+       description: project fields and filter
+     - source-table: adb.web_order02
+       projection: \*, format('%S', product_name) as product_name
+       filter: addone(id) > 20 AND order_id > 200
+       description: project fields and filter
+
+   route:
+     - source-table: app_db.orders
+       sink-table: ods_db.ods_orders
+     - source-table: app_db.shipments
+       sink-table: ods_db.ods_shipments
+     - source-table: app_db.products
+       sink-table: ods_db.ods_products
+
+   pipeline:
+     name: Sync MySQL Database to Doris
+     parallelism: 2
+     user-defined-function:
+       - name: addone
+         classpath: com.example.functions.AddOneFunctionClass
+       - name: format
+         classpath: com.example.functions.FormatFunctionClass
   ```
 4. Submit pipeline job using `flink-cdc.sh` script.
  ```shell

docs/config.toml

@@ -34,11 +34,11 @@ pygmentsUseClasses = true
   # we change the version for the complete docs when forking of a release branch
   # etc.
   # The full version string as referenced in Maven (e.g. 1.2.1)
-  Version = "3.2-SNAPSHOT"
+  Version = "3.3-SNAPSHOT"
 
   # For stable releases, leave the bugfix version out (e.g. 1.2). For snapshot
   # release this should be the same as the regular version
-  VersionTitle = "3.2-SNAPSHOT"
+  VersionTitle = "3.3-SNAPSHOT"
 
   # The branch for this version of Apache Flink CDC
   Branch = "master"
@@ -58,8 +58,8 @@ pygmentsUseClasses = true
   ]
 
   PreviousDocs = [
+    ["3.2", "https://nightlies.apache.org/flink/flink-cdc-docs-release-3.2"],
     ["3.1", "https://nightlies.apache.org/flink/flink-cdc-docs-release-3.1"],
-    ["3.0", "https://nightlies.apache.org/flink/flink-cdc-docs-release-3.0"],
   ]
 
 [markup]

docs/content.zh/docs/connectors/flink-sources/mongodb-cdc.md

@@ -39,7 +39,7 @@ MongoDB CDC 连接器允许从 MongoDB 读取快照数据和增量数据。 本
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-mongodb-cdc</artifactId>
    <!--  请使用已发布的版本依赖，snapshot 版本的依赖需要本地自行编译。 -->
-   <version>3.2-SNAPSHOT</version>
+   <version>3.3-SNAPSHOT</version>
 </dependency>
 ```
 

docs/content.zh/docs/connectors/flink-sources/mysql-cdc.md

@@ -46,7 +46,7 @@ MySQL CDC 连接器允许从 MySQL 数据库读取快照数据和增量数据。
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-mysql-cdc</artifactId>
    <!--  请使用已发布的版本依赖，snapshot 版本的依赖需要本地自行编译。 -->
-   <version>3.2-SNAPSHOT</version>
+   <version>3.3-SNAPSHOT</version>
 </dependency>
 ```
 

docs/content.zh/docs/connectors/flink-sources/tutorials/oceanbase-tutorial.md

@@ -42,16 +42,23 @@ under the License.
 version: '2.1'
 services:
   observer:
-    image: oceanbase/oceanbase-ce:4.0.0.0
+    image: 'oceanbase/oceanbase-ce:4.2.1.6-106000012024042515'
     container_name: observer
-    network_mode: "host"
+    environment:
+      - 'MODE=mini'
+      - 'OB_SYS_PASSWORD=123456'
+      - 'OB_TENANT_PASSWORD=654321'
+    ports:
+      - '2881:2881'
+      - '2882:2882'
   oblogproxy:
-    image: whhe/oblogproxy:1.1.0_4x
+    image: 'oceanbase/oblogproxy-ce:latest'
     container_name: oblogproxy
     environment:
       - 'OB_SYS_USERNAME=root'
-      - 'OB_SYS_PASSWORD=pswd'
-    network_mode: "host"
+      - 'OB_SYS_PASSWORD=123456'
+    ports:
+      - '2983:2983'
   elasticsearch:
     image: 'elastic/elasticsearch:7.6.0'
     container_name: elasticsearch
@@ -85,42 +92,26 @@ services:
 docker-compose up -d
 ```
 
-### 设置密码
-
-OceanBase 中 root 用户默认是没有密码的，但是 oblogproxy 需要配置一个使用非空密码的系统租户用户，因此这里我们需要先为 root@sys 用户设置一个密码。
+### 查询 Root Service List
 
 登陆 sys 租户的 root 用户：
 
 ```shell
-docker-compose exec observer obclient -h127.0.0.1 -P2881 -uroot@sys
-```
-
-设置密码，注意这里的密码需要与上一步中 oblogproxy 服务的环境变量 'OB_SYS_PASSWORD' 保持一样。
-
-```mysql
-ALTER USER root IDENTIFIED BY 'pswd';
-```
-
-OceanBase 从社区版 4.0.0.0 开始只支持对非 sys 租户的增量数据拉取，这里我们使用 test 租户的 root 用户作为示例。
-
-登陆 test 租户的 root 用户：
-
-```shell
-docker-compose exec observer obclient -h127.0.0.1 -P2881 -uroot@test
+docker-compose exec observer obclient -h127.0.0.1 -P2881 -uroot@sys -p123456
 ```
 
-设置密码:
+执行以下 sql 以查询 root service list，将 VALUE 列的值保存下来。
 
 ```mysql
-ALTER USER root IDENTIFIED BY 'test';
+SHOW PARAMETERS LIKE 'rootservice_list';
 ```
 
 ### 准备数据
 
-使用 'root@test' 用户登陆。
+使用测试用的 test 租户的 root 用户登陆。
 
 ```shell
-docker-compose exec observer obclient -h127.0.0.1 -P2881 -uroot@test -ptest
+docker-compose exec observer obclient -h127.0.0.1 -P2881 -uroot@test -p654321
 ```
 
 ```sql
@@ -169,6 +160,8 @@ VALUES (default, '2020-07-30 10:08:22', 'Jark', 50.50, 102, false),
 
 ### 在 Flink SQL CLI 中使用 Flink DDL 创建表
 
+注意在 OceanBase 源表的 SQL 中替换 root_service_list 为真实值。
+
 ```sql
 -- 设置间隔时间为3秒                       
 Flink SQL> SET execution.checkpointing.interval = 3s;
@@ -189,13 +182,13 @@ Flink SQL> CREATE TABLE orders (
     'connector' = 'oceanbase-cdc',
     'scan.startup.mode' = 'initial',
     'username' = 'root@test',
-    'password' = 'test',
+    'password' = '654321',
     'tenant-name' = 'test',
     'database-name' = '^ob$',
     'table-name' = '^orders$',
     'hostname' = 'localhost',
     'port' = '2881',
-    'rootserver-list' = '127.0.0.1:2882:2881',
+    'rootserver-list' = '${root_service_list}',
     'logproxy.host' = 'localhost',
     'logproxy.port' = '2983',
     'working-mode' = 'memory'
@@ -211,13 +204,13 @@ Flink SQL> CREATE TABLE products (
     'connector' = 'oceanbase-cdc',
     'scan.startup.mode' = 'initial',
     'username' = 'root@test',
-    'password' = 'test',
+    'password' = '654321',
     'tenant-name' = 'test',
     'database-name' = '^ob$',
     'table-name' = '^products$',
     'hostname' = 'localhost',
     'port' = '2881',
-    'rootserver-list' = '127.0.0.1:2882:2881',
+    'rootserver-list' = '${root_service_list}',
     'logproxy.host' = 'localhost',
     'logproxy.port' = '2983',
     'working-mode' = 'memory'

docs/content.zh/docs/connectors/pipeline-connectors/doris.md

@@ -162,7 +162,7 @@ pipeline:
       <td>String</td>
       <td>StreamLoad的参数。
         For example: <code> sink.properties.strict_mode: true</code>.
-        查看更多关于 <a href="https://doris.apache.org/zh-CN/docs/dev/sql-manual/sql-statements/Data-Manipulation-Statements/Load/STREAM-LOAD/"> StreamLoad的Properties 属性</a></td> 
+        查看更多关于 <a href="https://doris.apache.org/zh-CN/docs/dev/sql-manual/sql-statements/Data-Manipulation-Statements/Load/STREAM-LOAD/"> StreamLoad 的属性</a></td> 
       </td>
     </tr>
     <tr>
@@ -172,7 +172,7 @@ pipeline:
       <td>String</td>
       <td>创建表的Properties配置。
         For example: <code> table.create.properties.replication_num: 1</code>.
-        查看更多关于 <a href="https://doris.apache.org/zh-CN/docs/dev/sql-manual/sql-statements/Data-Definition-Statements/Create/CREATE-TABLE/"> Doris Table 的  Properties 属性</a></td> 
+        查看更多关于 <a href="https://doris.apache.org/zh-CN/docs/dev/sql-manual/sql-statements/Data-Definition-Statements/Create/CREATE-TABLE/"> Doris Table 的属性</a></td> 
       </td>
     </tr>
     </tbody>
@@ -186,7 +186,7 @@ pipeline:
     <thead>
       <tr>
         <th class="text-left" style="width:10%;">CDC type</th>
-        <th class="text-left" style="width:30%;">Doris type<a href="https://doris.apache.org/zh-CN/docs/dev/sql-manual/sql-types/Data-Types/BOOLEAN/"></a></th>
+        <th class="text-left" style="width:30%;">Doris type</th>
         <th class="text-left" style="width:60%;">NOTE</th>
       </tr>
     </thead>