Merge branch 'apache:master' into master

.github/ISSUE_TEMPLATE/bug-report.yml

@@ -76,6 +76,7 @@ body:
       multiple: false
       options:
         - 'master'
+        - '1.13.0'
         - '1.12.0'
         - '1.11.0'
         - '1.10.0'

.github/workflows/ci_ut.yml

@@ -101,7 +101,7 @@ jobs:
           CI: false
 
       - name: Unit test with Maven
-        run: mvn --batch-mode --update-snapshots -e -V test -pl !:sort-end-to-end-tests-v1.15,!:sort-end-to-end-tests-v1.13
+        run: mvn --batch-mode --update-snapshots -e -V test -pl !:sort-end-to-end-tests-v1.15,!:sort-end-to-end-tests-v1.13,!:sort-end-to-end-tests-v1.18
         env:
           CI: false
 

.github/workflows/ci_ut_flink18.yml

@@ -0,0 +1,96 @@
+#
+# 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.
+#
+
+name:
+  InLong Unit Test For Flink 1.18
+
+on:
+  push:
+    paths:
+      - '.github/workflows/ci_ut_flink18.yml'
+      - 'inlong-sort/**'
+      - '!**.md'
+
+  pull_request:
+    paths:
+      - '.github/workflows/ci_ut_flink18.yml'
+      - 'inlong-sort/**'
+      - '!**.md'
+
+jobs:
+  unit-test:
+    name: Unit Test
+    runs-on: ubuntu-22.04
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      # Release space size
+      - name: Remove unnecessary packages
+        run: |
+          echo "=== Before pruning ==="
+          df -h
+          sudo rm -rf /usr/share/dotnet
+          sudo rm -rf /usr/local/lib/android
+          sudo rm -rf /opt/ghc
+          sudo rm -rf /opt/hostedtoolcache
+          echo "=== After pruning ==="
+          df -h
+
+      - name: Set up JDK
+        uses: actions/setup-java@v4
+        with:
+          java-version: 8
+          distribution: adopt
+
+      - name: Cache Maven packages
+        uses: actions/cache@v4
+        with:
+          path: |
+            ~/.m2/repository/*/*/*
+            !~/.m2/repository/org/apache/inlong
+          key: ${{ runner.os }}-inlong-flink18-${{ hashFiles('**/pom.xml') }}
+          restore-keys: ${{ runner.os }}-inlong-flink18
+
+      - name: Build for Flink 1.18 with Maven
+        run: mvn --update-snapshots -e -V clean install -U -pl :sort-end-to-end-tests-v1.18 -am -Pv1.18 -DskipTests -Dhttp.keepAlive=false -Dmaven.wagon.http.pool=false -Dmaven.wagon.httpconnectionManager.ttlSeconds=120 -Daether.connector.http.reuseConnections=false -Daether.connector.requestTimeout=60000
+        env:
+          CI: false
+
+      - name: Unit test for Flink 1.18 with Maven
+        run: mvn --update-snapshots -e -V verify -pl :sort-end-to-end-tests-v1.18 -am -Pv1.18
+        env:
+          CI: false
+
+      - name: Upload unit test results
+        if: ${{ failure() }}
+        uses: actions/upload-artifact@v4
+        with:
+          name: surefire-reports
+          path: ./**/target/surefire-reports/
+          if-no-files-found: ignore
+
+      - name: Upload integration test results
+        if: ${{ failure() }}
+        uses: actions/upload-artifact@v4
+        with:
+          name: failsafe-reports
+          path: ./**/target/failsafe-reports/
+          if-no-files-found: ignore
+
+      - name: Clean up build packages
+        run: mvn clean

README.md

@@ -104,6 +104,7 @@ You can use InLong in the following ways：
 |              | StarRocks         | >= 2.0                       |
 |              | Kudu              | >= 1.12.0                    |
 |              | Redis             | >= 3.0                       |
+|              | OceanBase         | >= 1.0                       |
 
 ## Build InLong
 More detailed instructions can be found at [Quick Start](https://inlong.apache.org/docs/next/quick_start/how_to_build) section in the documentation.

bin/init-config.sh

@@ -51,19 +51,17 @@ detect_sed_command
 init_inlong_agent() {
   echo "Init agent configuration parameters"
   cd $INLONG_HOME/inlong-agent/conf
-  $SED_COMMAND 's/agent.local.ip=.*/'''agent.local.ip=${local_ip}'''/g' agent.properties
-  $SED_COMMAND 's/agent.http.port=.*/'''agent.http.port=${agent_port}'''/g' agent.properties
-  $SED_COMMAND 's/agent.manager.addr=.*/'''agent.manager.addr=http://${manager_server_hostname}:${manager_server_port}'''/g' agent.properties
-  $SED_COMMAND "s/audit.enable=.*$/audit.enable=true/g" agent.properties
-  $SED_COMMAND 's/audit.proxys=.*/'''audit.proxys=${audit_proxys_ip}:${audit_proxys_port}'''/g' agent.properties
+  $SED_COMMAND "s|agent.local.ip=.*|agent.local.ip=${local_ip}|g" agent.properties
+  $SED_COMMAND "s|agent.manager.addr=.*|agent.manager.addr=http://${manager_server_hostname}:${manager_server_port}|g" agent.properties
 }
 
 init_inlong_audit() {
   echo "Init audit configuration parameters"
   cd $INLONG_HOME/inlong-audit/conf
+  # configure Audit Store
   $SED_COMMAND 's#jdbc:mysql://.*apache_inlong_audit#'''jdbc:mysql://${spring_datasource_hostname}:${spring_datasource_port}/apache_inlong_audit'''#g' application.properties
-  $SED_COMMAND 's/spring.datasource.druid.username=.*/'''spring.datasource.druid.username=${spring_datasource_username}'''/g' application.properties
-  $SED_COMMAND 's/spring.datasource.druid.password=.*/'''spring.datasource.druid.password=${spring_datasource_password}'''/g' application.properties
+  $SED_COMMAND 's/jdbc.username=.*/'''jdbc.username=${spring_datasource_username}'''/g' application.properties
+  $SED_COMMAND 's/jdbc.password=.*/'''jdbc.password=${spring_datasource_password}'''/g' application.properties
   if [ $mq_type == "pulsar" ]; then
     $SED_COMMAND 's/audit.config.proxy.type=.*/'''audit.config.proxy.type=pulsar'''/g' application.properties
   fi
@@ -73,14 +71,20 @@ init_inlong_audit() {
   if [ $mq_type == "tubemq" ]; then
     $SED_COMMAND 's/audit.config.proxy.type=.*/'''audit.config.proxy.type=tube'''/g' application.properties
   fi
+  # configure Audit Service
+  $SED_COMMAND 's#jdbc:mysql://.*apache_inlong_audit#'''jdbc:mysql://${spring_datasource_hostname}:${spring_datasource_port}/apache_inlong_audit'''#g' audit-service.properties
+  $SED_COMMAND 's/mysql.username=.*/'''mysql.username=${spring_datasource_username}'''/g' audit-service.properties
+  $SED_COMMAND 's/mysql.password=.*/'''mysql.password=${spring_datasource_password}'''/g' audit-service.properties
+  $SED_COMMAND 's/audit.proxy.address.agent=.*/'''audit.proxy.address.agent=${audit_service_ip}:${audit_proxy_port}'''/g' audit-service.properties
+  $SED_COMMAND 's/audit.proxy.address.dataproxy=.*/'''audit.proxy.address.dataproxy=${audit_service_ip}:${audit_proxy_port}'''/g' audit-service.properties
+  $SED_COMMAND 's/audit.proxy.address.sort=.*/'''audit.proxy.address.sort=${audit_service_ip}:${audit_proxy_port}'''/g' audit-service.properties
 }
 
 init_inlong_dataproxy() {
   echo "Init dataproxy configuration parameters"
   cd $INLONG_HOME/inlong-dataproxy/conf
   $SED_COMMAND 's/manager.hosts=.*/'''manager.hosts=${manager_server_hostname}:${manager_server_port}'''/g' common.properties
-  $SED_COMMAND 's/audit.proxys=.*/'''audit.proxys=${audit_proxys_ip}:${audit_proxys_port}'''/g' common.properties
-  $SED_COMMAND "s/audit.enable=.*$/audit.enable=true/g" common.properties
+  $SED_COMMAND "s/audit.proxys.discovery.manager.enable=.*$/audit.proxys.discovery.manager.enable=true/g" common.properties
 }
 
 init_inlong_manager() {
@@ -92,11 +96,13 @@ init_inlong_manager() {
     $SED_COMMAND 's#jdbc:mysql://.*apache_inlong_manager#'''jdbc:mysql://${spring_datasource_hostname}:${spring_datasource_port}/apache_inlong_manager'''#g' application-dev.properties
     $SED_COMMAND 's/spring.datasource.druid.username=.*/'''spring.datasource.druid.username=${spring_datasource_username}'''/g' application-dev.properties
     $SED_COMMAND 's/spring.datasource.druid.password=.*/'''spring.datasource.druid.password=${spring_datasource_password}'''/g' application-dev.properties
+    $SED_COMMAND 's/audit.query.url=.*/'''audit.query.url=${audit_service_ip}:${audit_service_port}'''/g' application-dev.properties
   fi
   if [ $spring_profiles_active == "prod" ]; then
     $SED_COMMAND 's#jdbc:mysql://.*apache_inlong_manager#'''jdbc:mysql://${spring_datasource_hostname}:${spring_datasource_port}/apache_inlong_manager'''#g' application-prod.properties
     $SED_COMMAND 's/spring.datasource.druid.username=.*/'''spring.datasource.druid.username=${spring_datasource_username}'''/g' application-prod.properties
     $SED_COMMAND 's/spring.datasource.druid.password=.*/'''spring.datasource.druid.password=${spring_datasource_password}'''/g' application-prod.properties
+    $SED_COMMAND 's/audit.query.url=.*/'''audit.query.url=${audit_service_ip}:${audit_service_port}'''/g' application-prod.properties
   fi
   echo "Init inlong manager flink plugin configuration"
   cd $INLONG_HOME/inlong-manager/plugins

bin/inlong-daemon

@@ -83,10 +83,14 @@ start_inlong_audit() {
   if [ "${mq_type}" = "kafka" ]; then
     bash +x ./bin/proxy-start.sh kafka
   fi
+  wait_port_to_listen audit ${audit_proxy_port}
   echo "start audit store"
   bash +x ./bin/store-start.sh
   # wait to start
-  wait_port_to_listen audit ${audit_proxys_port}
+  echo "start audit service"
+  bash +x ./bin/service-start.sh
+  # wait to start
+  wait_port_to_listen audit ${audit_service_port}
 }
 
 start_inlong_manager() {
@@ -136,8 +140,6 @@ start_inlong_agent() {
   echo "start agent"
   cd $INLONG_HOME/inlong-agent
   bash +x ./bin/agent.sh start
-  # wait to start
-  wait_port_to_listen agent ${agent_port}
 }
 
 # start inlong
@@ -179,6 +181,7 @@ stop_inlong_audit() {
   cd $INLONG_HOME/inlong-audit/bin
   bash +x ./proxy-stop.sh
   bash +x ./store-stop.sh
+  bash +x ./service-stop.sh
 }
 
 stop_inlong_agent() {

conf/inlong.conf

@@ -48,11 +48,10 @@ flink_rest_port=8081
 ############## DataProxy Configuration ##############
 dataproxy_port=46801
 
-############## Agent Configuration ##############
-agent_port=8008
-
 ############## Audit Configuration ##############
-# audit proxy IP
-audit_proxys_ip=127.0.0.1
+# audit service IP
+audit_service_ip=127.0.0.1
+# audit service Port
+audit_service_port=10080
 # audit proxy Port
-audit_proxys_port=10081
+audit_proxy_port=10081

docker/docker-compose/docker-compose.yml

@@ -33,6 +33,7 @@ services:
     volumes:
       - ./mysql:/var/lib/mysql
       - ./sql:/docker-entrypoint-initdb.d
+    command: --authentication_policy=mysql_native_password
 
   pulsar:
     image: apachepulsar/pulsar:2.8.2
@@ -59,7 +60,7 @@ services:
       - ZK_URL=tubemq-server:2181
       - FLINK_HOST=jobmanager
       - FLINK_PORT=8081
-      - AUDIT_PROXY_URL=audit:10081
+      - AUDIT_QUERY_URL=http://audit:10080
 
   dashboard:
     image: inlong/dashboard:${VERSION_TAG}
@@ -82,7 +83,6 @@ services:
     environment:
       - MANAGER_OPENAPI_IP=manager
       - MANAGER_OPENAPI_PORT=8083
-      - AUDIT_PROXY_URL=audit:10081
       # pulsar or kafka
       - MQ_TYPE=pulsar
       - ETH_NAME=eth0
@@ -98,9 +98,6 @@ services:
       - MANAGER_OPENAPI_PORT=8083
       - DATAPROXY_IP=dataproxy
       - DATAPROXY_PORT=46801
-      - AUDIT_PROXY_URL=audit:10081
-    ports:
-      - "8008:8008"
     volumes:
       - ./collect-data:/data/collect-data
 
@@ -112,19 +109,21 @@ services:
       mysql:
         condition: service_healthy
     environment:
-      - JDBC_URL=mysql:3306
-      - USERNAME=root
-      - PASSWORD=inlong
+      - AUDIT_JDBC_URL=mysql:3306
+      - AUDIT_JDBC_USERNAME=root
+      - AUDIT_JDBC_PASSWORD=inlong
       - MANAGER_OPENAPI_IP=manager
       - MANAGER_OPENAPI_PORT=8083
+      - AUDIT_PROXY_ADDRESS=audit:10081
       # pulsar or kafka
       - MQ_TYPE=pulsar
     ports:
+      - "10080:10080"
       - "10081:10081"
 
   # flink jobmanager
   jobmanager:
-    image: apache/flink:1.13-scala_2.11
+    image: apache/flink:1.15-scala_2.12
     container_name: jobmanager
     environment:
       - |
@@ -136,7 +135,7 @@ services:
 
   # flink taskmanager
   taskmanager:
-    image: apache/flink:1.13-scala_2.11
+    image: apache/flink:1.15-scala_2.12
     container_name: taskmanager
     environment:
       - |

inlong-agent/agent-common/pom.xml

@@ -23,7 +23,7 @@
     <parent>
         <groupId>org.apache.inlong</groupId>
         <artifactId>inlong-agent</artifactId>
-        <version>1.13.0-SNAPSHOT</version>
+        <version>1.14.0-SNAPSHOT</version>
     </parent>
 
     <artifactId>agent-common</artifactId>

inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/conf/InstanceProfile.java

@@ -80,6 +80,10 @@ public String getInstanceId() {
         return get(TaskConstants.INSTANCE_ID);
     }
 
+    public String getCycleUnit() {
+        return get(TaskConstants.TASK_CYCLE_UNIT);
+    }
+
     public String getSourceClass() {
         return get(TaskConstants.TASK_SOURCE);
     }

inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constant/AgentConstants.java

@@ -31,30 +31,32 @@ public class AgentConstants {
     public static final String DEFAULT_AGENT_HOME = System.getProperty("agent.home");
     public static final String AGENT_ROCKS_DB_PATH = "agent.rocks.db.path";
     public static final String DEFAULT_AGENT_ROCKS_DB_PATH = ".rocksdb";
-    public static final String AGENT_LOCAL_DB_PATH_TASK = ".localdb/task";
-    public static final String AGENT_LOCAL_DB_PATH_INSTANCE = ".localdb/instance";
-    public static final String AGENT_LOCAL_DB_PATH_OFFSET = ".localdb/offset";
+    public static final String AGENT_STORE_PATH_TASK = ".localdb/task";
+    public static final String AGENT_STORE_PATH_INSTANCE = ".localdb/instance";
+    public static final String AGENT_STORE_PATH_OFFSET = ".localdb/offset";
     public static final String AGENT_UNIQ_ID = "agent.uniq.id";
     // default is empty.
     public static final String AGENT_FETCHER_CLASSNAME = "agent.fetcher.classname";
     public static final String AGENT_CONF_PARENT = "agent.conf.parent";
     public static final String DEFAULT_AGENT_CONF_PARENT = "conf";
-    public static final String AGENT_HTTP_PORT = "agent.http.port";
-    public static final int DEFAULT_AGENT_HTTP_PORT = 8008;
     public static final String CHANNEL_MEMORY_CAPACITY = "channel.memory.capacity";
     public static final int DEFAULT_CHANNEL_MEMORY_CAPACITY = 2000;
     public static final String JOB_NUMBER_LIMIT = "job.number.limit";
     public static final int DEFAULT_JOB_NUMBER_LIMIT = 15;
     public static final String AGENT_LOCAL_IP = "agent.local.ip";
     public static final String DEFAULT_LOCAL_IP = "127.0.0.1";
     public static final String DEFAULT_LOCAL_HOST = "localhost";
+    public static final String AGENT_STORE_CLASSNAME = "agent.store.classname";
+    public static final String DEFAULT_AGENT_STORE_CLASSNAME = "org.apache.inlong.agent.plugin.store.RocksDBStoreImpl";
+
     // default use local ip as uniq id for agent.
     public static final String DEFAULT_AGENT_UNIQ_ID = AgentUtils.getLocalIp();
     public static final String CUSTOM_FIXED_IP = "agent.custom.fixed.ip";
 
     public static final String AGENT_CLUSTER_NAME = "agent.cluster.name";
     public static final String AGENT_CLUSTER_TAG = "agent.cluster.tag";
     public static final String AGENT_CLUSTER_IN_CHARGES = "agent.cluster.inCharges";
+    public static final String AGENT_INSTALL_PLATFORM = "agent.install.platform";
 
     public static final String AGENT_LOCAL_UUID = "agent.local.uuid";
     public static final String AGENT_LOCAL_UUID_OPEN = "agent.local.uuid.open";
@@ -67,9 +69,6 @@ public class AgentConstants {
     public static final String AUDIT_ENABLE = "audit.enable";
     public static final boolean DEFAULT_AUDIT_ENABLE = true;
 
-    public static final String AUDIT_KEY_PROXYS = "audit.proxys";
-    public static final String DEFAULT_AUDIT_PROXYS = "";
-
     public static final String AGENT_HISTORY_PATH = "agent.history.path";
     public static final String DEFAULT_AGENT_HISTORY_PATH = ".history";
 

inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constant/CommonConstants.java

@@ -89,9 +89,9 @@ public class CommonConstants {
     public static final String PROXY_KEY_DATA = "dataKey";
 
     public static final int DEFAULT_FILE_MAX_NUM = 4096;
-    public static final String TASK_ID_PREFIX = "task_";
-    public static final String INSTANCE_ID_PREFIX = "ins_";
-    public static final String OFFSET_ID_PREFIX = "offset_";
+    public static final String TASK_ID_PREFIX = "task";
+    public static final String INSTANCE_ID_PREFIX = "ins";
+    public static final String OFFSET_ID_PREFIX = "offset";
     public static final String AGENT_OS_NAME = "os.name";
     public static final String AGENT_NIX_OS = "nix";
     public static final String AGENT_NUX_OS = "nux";

inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constant/CycleUnitType.java

@@ -19,7 +19,8 @@
 
 public class CycleUnitType {
 
-    public static final String DAY = "D";
-    public static final String HOUR = "h";
+    public static final String DAY = "d";
+    public static final String HOUR = "H";
+    public static final String MINUTE = "m";
     public static final String REAL_TIME = "R";
 }