[Improve] Support retry on lookup join

.github/workflows/build-connector.yml

@@ -40,42 +40,24 @@ jobs:
 
     - name: Build flink connector 1.15
       run: |
-        cd flink-doris-connector && mvn clean package \
-          -Dflink.version=1.15.0 \
-          -Dflink.minor.version=1.15 \
-          -Dflink.python.id=flink-python_2.12
+        cd flink-doris-connector && mvn clean install -pl flink-doris-connector-1.15 -am
 
     - name: Build flink connector 1.16
       run: |
-        cd flink-doris-connector && mvn clean package \
-          -Dflink.version=1.16.0 \
-          -Dflink.minor.version=1.16 \
-          -Dflink.python.id=flink-python
+        cd flink-doris-connector && mvn clean install -pl flink-doris-connector-1.16 -am
 
     - name: Build flink connector 1.17
       run: |
-        cd flink-doris-connector && mvn clean package \
-          -Dflink.version=1.17.0 \
-          -Dflink.minor.version=1.17 \
-          -Dflink.python.id=flink-python
+        cd flink-doris-connector && mvn clean install -pl flink-doris-connector-1.17 -am
 
     - name: Build flink connector 1.18
       run: |
-        cd flink-doris-connector && mvn clean package \
-          -Dflink.version=1.18.0 \
-          -Dflink.minor.version=1.18 \
-          -Dflink.python.id=flink-python
+        cd flink-doris-connector && mvn clean install -pl flink-doris-connector-1.18 -am
 
     - name: Build flink connector 1.19
       run: |
-        cd flink-doris-connector && mvn clean package \
-          -Dflink.version=1.19.0 \
-          -Dflink.minor.version=1.19 \
-          -Dflink.python.id=flink-python
+        cd flink-doris-connector && mvn clean install -pl flink-doris-connector-1.19 -am
 
     - name: Build flink connector 1.20
       run: |
-        cd flink-doris-connector && mvn clean package \
-          -Dflink.version=1.20.0 \
-          -Dflink.minor.version=1.20 \
-          -Dflink.python.id=flink-python
+        cd flink-doris-connector && mvn clean install -pl flink-doris-connector-1.20 -am

.github/workflows/checkstyle.yaml

@@ -40,4 +40,4 @@ jobs:
 
       - name: Run java checkstyle
         run:
-          cd flink-doris-connector && mvn clean compile checkstyle:checkstyle
+          cd flink-doris-connector && mvn clean install -DskipTests && mvn checkstyle:checkstyle

.github/workflows/run-e2ecase.yml

@@ -40,5 +40,5 @@ jobs:
 
     - name: Run E2ECases
       run: |
-        cd flink-doris-connector && mvn test -Dtest="*E2ECase" -Dimage="apache/doris:doris-all-in-one-2.1.0"
+        cd flink-doris-connector && mvn clean test -DfailIfNoTests=false -Dtest="*E2ECase" -Dimage="apache/doris:doris-all-in-one-2.1.0"
 

.github/workflows/run-itcase.yml

@@ -38,7 +38,11 @@ jobs:
         distribution: adopt
         java-version: '8'
 
-    - name: Run ITCases
+    - name: Run ITCases For 1.15
       run: |
-        cd flink-doris-connector && mvn test -Dtest="*ITCase" -Dimage="apache/doris:doris-all-in-one-2.1.0"
+        cd flink-doris-connector && mvn clean test -pl flink-doris-connector-1.15 -am -DfailIfNoTests=false -Dtest="*ITCase" -Dimage="apache/doris:doris-all-in-one-2.1.0"
+
+    - name: Run ITCases For 1.18
+      run: |
+        cd flink-doris-connector && mvn clean test -pl flink-doris-connector-1.18 -am -DfailIfNoTests=false -Dtest="*ITCase" -Dimage="apache/doris:doris-all-in-one-2.1.0"
 

.licenserc.yaml

@@ -12,6 +12,6 @@ header:
     - '.github/PULL_REQUEST_TEMPLATE.md'
     - '.licenserc.yaml'
     - 'custom_env.sh.tpl'
-    - 'flink-doris-connector/src/test/resources/container/'
+    - 'flink-doris-connector/flink-doris-connector-common/src/test/resources/container/'
 
   comment: on-failure

flink-doris-connector/build.sh

@@ -151,7 +151,6 @@ flinkVer=$?
 FLINK_PYTHON_ID="flink-python"
 if [ ${flinkVer} -eq 1 ]; then
     FLINK_VERSION="1.15.0"
-    FLINK_PYTHON_ID="flink-python_2.12"
 elif [ ${flinkVer} -eq 2 ]; then
     FLINK_VERSION="1.16.0"
 elif [ ${flinkVer} -eq 3 ]; then
@@ -172,7 +171,7 @@ FLINK_MAJOR_VERSION=0
 echo_g " flink version: ${FLINK_VERSION}, major version: ${FLINK_MAJOR_VERSION}"
 echo_g " build starting..."
 
-${MVN_BIN} clean package -Dflink.version=${FLINK_VERSION} -Dflink.major.version=${FLINK_MAJOR_VERSION} -Dflink.python.id=${FLINK_PYTHON_ID} "$@"
+${MVN_BIN} clean install -pl flink-doris-connector-${FLINK_MAJOR_VERSION} -am "$@"
 
 EXIT_CODE=$?
 if [ $EXIT_CODE -eq 0 ]; then

flink-doris-connector/flink-doris-connector-1.15/pom.xml

@@ -0,0 +1,52 @@
+<?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/maven-v4_0_0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.apache.doris</groupId>
+        <artifactId>flink-doris-connector</artifactId>
+        <version>25.0.0-SNAPSHOT</version>
+    </parent>
+    <artifactId>flink-doris-connector-1.15</artifactId>
+    <name>Flink Doris Connector Flink 1.15</name>
+
+    <properties>
+        <flink.version>1.15.3</flink.version>
+        <flink.major.version>1.15</flink.major.version>
+        <flink.python.id>flink-python_2.12</flink.python.id>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.doris</groupId>
+            <artifactId>flink-doris-connector-common</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.doris</groupId>
+            <artifactId>flink-doris-connector-common</artifactId>
+            <version>${project.version}</version>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+</project>

flink-doris-connector/flink-doris-connector-1.15/src/main/java/org/apache/doris/flink/lookup/LookupRuntimeProviderFactory.java

@@ -0,0 +1,35 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.flink.lookup;
+
+import org.apache.flink.table.connector.source.AsyncTableFunctionProvider;
+import org.apache.flink.table.connector.source.LookupTableSource;
+import org.apache.flink.table.connector.source.TableFunctionProvider;
+import org.apache.flink.table.functions.AsyncTableFunction;
+import org.apache.flink.table.functions.TableFunction;
+
+/** Factory to create {@link LookupTableSource.LookupRuntimeProvider}. */
+public class LookupRuntimeProviderFactory {
+
+    public static LookupTableSource.LookupRuntimeProvider create(
+            DorisLookupFunction function, boolean enableAsync) {
+        return enableAsync
+                ? AsyncTableFunctionProvider.of((AsyncTableFunction) function)
+                : TableFunctionProvider.of((TableFunction) function);
+    }
+}

flink-doris-connector/src/main/java/org/apache/doris/flink/table/DorisRowDataAsyncLookupFunction.java → flink-doris-connector/flink-doris-connector-1.15/src/main/java/org/apache/doris/flink/table/DorisRowDataAsyncLookupFunction.java

@@ -30,6 +30,7 @@
 import org.apache.doris.flink.cfg.DorisLookupOptions;
 import org.apache.doris.flink.cfg.DorisOptions;
 import org.apache.doris.flink.lookup.DorisJdbcLookupReader;
+import org.apache.doris.flink.lookup.DorisLookupFunction;
 import org.apache.doris.flink.lookup.DorisLookupReader;
 import org.apache.doris.flink.lookup.LookupMetrics;
 import org.apache.doris.flink.lookup.LookupSchema;
@@ -39,18 +40,19 @@
 import java.io.IOException;
 import java.util.Collection;
 import java.util.Collections;
-import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 
-public class DorisRowDataAsyncLookupFunction extends AsyncTableFunction<RowData> {
+/** async lookup {@link AsyncTableFunction} for 1.15. */
+public class DorisRowDataAsyncLookupFunction extends AsyncTableFunction<RowData>
+        implements DorisLookupFunction {
     private static final Logger LOG =
             LoggerFactory.getLogger(DorisRowDataAsyncLookupFunction.class);
     private final DorisOptions options;
     private final DorisLookupOptions lookupOptions;
     private final long cacheMaxSize;
     private final long cacheExpireMs;
-    private transient Cache<RowData, List<RowData>> cache;
+    private transient Cache<RowData, Collection<RowData>> cache;
     private DorisLookupReader lookupReader;
     private LookupSchema lookupSchema;
     private LookupMetrics lookupMetrics;
@@ -101,7 +103,7 @@ public void eval(CompletableFuture<Collection<RowData>> future, Object... keys)
             throws IOException {
         RowData keyRow = GenericRowData.of(keys);
         if (cache != null) {
-            List<RowData> cachedRows = cache.getIfPresent(keyRow);
+            Collection<RowData> cachedRows = cache.getIfPresent(keyRow);
             if (cachedRows != null) {
                 lookupMetrics.incHitCount();
                 if (LOG.isDebugEnabled()) {
@@ -114,7 +116,7 @@ public void eval(CompletableFuture<Collection<RowData>> future, Object... keys)
             }
         }
 
-        CompletableFuture<List<RowData>> resultFuture = lookupReader.asyncGet(keyRow);
+        CompletableFuture<Collection<RowData>> resultFuture = lookupReader.asyncGet(keyRow);
         resultFuture.handleAsync(
                 (resultRows, throwable) -> {
                     try {
@@ -151,7 +153,7 @@ public void close() throws Exception {
     }
 
     @VisibleForTesting
-    public Cache<RowData, List<RowData>> getCache() {
+    public Cache<RowData, Collection<RowData>> getCache() {
         return cache;
     }
 }

flink-doris-connector/src/main/java/org/apache/doris/flink/table/DorisRowDataJdbcLookupFunction.java → flink-doris-connector/flink-doris-connector-1.15/src/main/java/org/apache/doris/flink/table/DorisRowDataJdbcLookupFunction.java

@@ -30,25 +30,27 @@
 import org.apache.doris.flink.cfg.DorisLookupOptions;
 import org.apache.doris.flink.cfg.DorisOptions;
 import org.apache.doris.flink.lookup.DorisJdbcLookupReader;
+import org.apache.doris.flink.lookup.DorisLookupFunction;
 import org.apache.doris.flink.lookup.DorisLookupReader;
 import org.apache.doris.flink.lookup.LookupMetrics;
 import org.apache.doris.flink.lookup.LookupSchema;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
+import java.util.Collection;
 import java.util.Collections;
-import java.util.List;
 import java.util.concurrent.TimeUnit;
 
-/** use jdbc to query. */
-public class DorisRowDataJdbcLookupFunction extends TableFunction<RowData> {
+/** lookup {@link TableFunction} for 1.15. */
+public class DorisRowDataJdbcLookupFunction extends TableFunction<RowData>
+        implements DorisLookupFunction {
     private static final Logger LOG = LoggerFactory.getLogger(DorisRowDataJdbcLookupFunction.class);
     private final DorisOptions options;
     private final DorisLookupOptions lookupOptions;
     private final long cacheMaxSize;
     private final long cacheExpireMs;
-    private transient Cache<RowData, List<RowData>> cache;
+    private transient Cache<RowData, Collection<RowData>> cache;
     private DorisLookupReader lookupReader;
     private LookupSchema lookupSchema;
     private LookupMetrics lookupMetrics;
@@ -97,7 +99,7 @@ public void open(FunctionContext context) throws Exception {
     public void eval(Object... keys) throws IOException {
         RowData keyRow = GenericRowData.of(keys);
         if (cache != null) {
-            List<RowData> cachedRows = cache.getIfPresent(keyRow);
+            Collection<RowData> cachedRows = cache.getIfPresent(keyRow);
             if (cachedRows != null) {
                 lookupMetrics.incHitCount();
                 if (LOG.isDebugEnabled()) {
@@ -115,7 +117,7 @@ public void eval(Object... keys) throws IOException {
     }
 
     private void queryRecord(RowData keyRow) throws IOException {
-        List<RowData> rowData = lookupReader.get(keyRow);
+        Collection<RowData> rowData = lookupReader.get(keyRow);
         if (rowData == null) {
             rowData = Collections.emptyList();
         }
@@ -135,7 +137,7 @@ public void close() throws Exception {
     }
 
     @VisibleForTesting
-    public Cache<RowData, List<RowData>> getCache() {
+    public Cache<RowData, Collection<RowData>> getCache() {
         return cache;
     }
 }