fix

JNSimba · JNSimba · commit 3b1dd102f92c · 2026-02-28T11:06:51.000+08:00
diff --git a/flink-doris-connector/flink-doris-connector-base/pom.xml b/flink-doris-connector/flink-doris-connector-base/pom.xml
@@ -29,7 +29,7 @@ under the License.
 
     <artifactId>flink-doris-connector-base</artifactId>
     <name>Flink Doris Connector Base</name>
-    <description>Shared code for Flink Doris Connector; built per Flink version via profile (flink1 / flink2)</description>
+    <description>Base code for Flink Doris Connector; built per Flink version via profile (flink1 / flink2)</description>
     <packaging>jar</packaging>
 
     <dependencies>
@@ -38,7 +38,6 @@ under the License.
             <artifactId>thrift-service</artifactId>
         </dependency>
 
-        <!-- Flink APIs needed by source/sink/catalog; version driven by profile -->
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-table-planner-loader</artifactId>
diff --git a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/sink/batch/DorisBatchWriter.java b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/sink/batch/DorisBatchWriter.java
@@ -47,8 +47,8 @@
  *
  * <p>Core batch stream load writer shared across Flink versions.
  *
- * <p>Version-specific modules should wrap this class (for example via DorisBatchWriterV1 /
- * DorisBatchWriterV2) and implement the appropriate Flink sink interfaces there.
+ * <p>Version-specific modules should wrap this class (for example via DorisBatchWriterAdapter and
+ * implement the appropriate Flink sink interfaces there.
  */
 public class DorisBatchWriter<IN> {
     private static final Logger LOG = LoggerFactory.getLogger(DorisBatchWriter.class);
diff --git a/flink-doris-connector/flink-doris-connector-flink1/pom.xml b/flink-doris-connector/flink-doris-connector-flink1/pom.xml
@@ -60,7 +60,7 @@ under the License.
             <artifactId>flink-connector-base</artifactId>
         </dependency>
 
-        <!-- Flink CDC (only in 1.20 module) -->
+        <!-- Flink CDC (only in flink1 module) -->
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-sql-connector-mysql-cdc</artifactId>
diff --git a/flink-doris-connector/flink-doris-connector-flink1/src/main/java/org/apache/doris/flink/sink/batch/DorisBatchWriterAdapter.java b/flink-doris-connector/flink-doris-connector-flink1/src/main/java/org/apache/doris/flink/sink/batch/DorisBatchWriterAdapter.java
@@ -32,7 +32,7 @@
 import java.util.Collection;
 import java.util.List;
 
-/** Flink 1.20 specific wrapper for the shared core {@link DorisBatchWriter} implementation. */
+/** Flink 1.x specific wrapper for the shared core {@link DorisBatchWriter} implementation. */
 public class DorisBatchWriterAdapter<IN>
         implements DorisAbstractWriter<IN, DorisWriterState, DorisCommittable> {
 
diff --git a/flink-doris-connector/flink-doris-connector-flink1/src/main/java/org/apache/doris/flink/sink/copy/DorisCopyWriterAdapter.java b/flink-doris-connector/flink-doris-connector-flink1/src/main/java/org/apache/doris/flink/sink/copy/DorisCopyWriterAdapter.java
@@ -31,7 +31,7 @@
 import java.util.Collection;
 import java.util.List;
 
-/** Flink 1.20 specific wrapper for the shared core {@link DorisCopyWriter} implementation. */
+/** Flink 1.x specific wrapper for the shared core {@link DorisCopyWriter} implementation. */
 public class DorisCopyWriterAdapter<IN>
         implements DorisAbstractWriter<IN, DorisWriterState, DorisCopyCommittable> {
 
diff --git a/flink-doris-connector/flink-doris-connector-flink1/src/main/java/org/apache/doris/flink/sink/writer/DorisWriterAdapter.java b/flink-doris-connector/flink-doris-connector-flink1/src/main/java/org/apache/doris/flink/sink/writer/DorisWriterAdapter.java
@@ -32,10 +32,10 @@
 import java.util.List;
 
 /**
- * Flink 1.20 specific wrapper for the shared core {@link DorisWriter} implementation.
+ * Flink 1.x specific wrapper for the shared core {@link DorisWriter} implementation.
  *
- * <p>This class adapts Flink 1.20's {@link Sink.InitContext} to the version-neutral core writer in
- * the base module while implementing {@link DorisAbstractWriter} for use by the 1.20 connector
+ * <p>This class adapts Flink 1.x's {@link Sink.InitContext} to the version-neutral core writer in
+ * the base module while implementing {@link DorisAbstractWriter} for use by the 1.x connector
  * module.
  */
 public class DorisWriterAdapter<IN>
diff --git a/flink-doris-connector/flink-doris-connector-flink1/src/test/java/org/apache/doris/flink/sink/DorisSinkTest.java b/flink-doris-connector/flink-doris-connector-flink1/src/test/java/org/apache/doris/flink/sink/DorisSinkTest.java
@@ -46,7 +46,7 @@ public class DorisSinkTest {
     private MockedStatic<BackendUtil> backendUtilMockedStatic;
 
     @Before
-    public void setUp() throws Exception{
+    public void setUp() throws Exception {
         backendUtilMockedStatic = mockStatic(BackendUtil.class);
         backendUtilMockedStatic.when(() -> BackendUtil.tryHttpConnection(any())).thenReturn(true);
     }
diff --git a/flink-doris-connector/flink-doris-connector-flink2/src/main/resources/MockSource.java b/flink-doris-connector/flink-doris-connector-flink2/src/main/resources/MockSource.java
diff --git a/flink-doris-connector/flink-doris-connector-it/src/test/java/org/apache/doris/flink/example/DataGen2DorisExample.java b/flink-doris-connector/flink-doris-connector-it/src/test/java/org/apache/doris/flink/example/DataGen2DorisExample.java
@@ -51,7 +51,7 @@ public static void main(String[] args) {
                         + "    ) "
                         + "    WITH ("
                         + "      'connector' = 'doris',"
-                        + "      'fenodes' = '10.16.10.6:28737',"
+                        + "      'fenodes' = '127.0.0.1:8030',"
                         + "      'table.identifier' = 'test.student',"
                         + "      'username' = 'root',"
                         + "      'password' = '',"
diff --git a/flink-doris-connector/flink-doris-connector-it/src/test/java/org/apache/doris/flink/example/DorisSourceExample.java b/flink-doris-connector/flink-doris-connector-it/src/test/java/org/apache/doris/flink/example/DorisSourceExample.java
@@ -39,7 +39,7 @@ public static void main(String[] args) throws Exception {
                         + "    ) "
                         + "    WITH ("
                         + "      'connector' = 'doris',"
-                        + "      'fenodes' = '10.16.10.6:28737',"
+                        + "      'fenodes' = '127.0.0.1:8030',"
                         + "      'table.identifier' = 'test.student',"
                         + "      'username' = 'root',"
                         + "      'password' = '',"
diff --git a/flink-doris-connector/flink-doris-connector-it/src/test/java/org/apache/doris/flink/source/DorisSourceITCase.java b/flink-doris-connector/flink-doris-connector-it/src/test/java/org/apache/doris/flink/source/DorisSourceITCase.java
@@ -108,32 +108,6 @@ public static Object[] parameters() {
                             .withHaLeadershipControl()
                             .build());
 
-    @Test
-    public void testOldSourceApi() throws Exception {
-        initializeTable(TABLE_READ_OLD_API, DataModel.UNIQUE);
-        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
-        env.setParallelism(DEFAULT_PARALLELISM);
-        Properties properties = new Properties();
-        properties.put("fenodes", getFenodes());
-        properties.put("username", getDorisUsername());
-        properties.put("password", getDorisPassword());
-        properties.put("table.identifier", DATABASE + "." + TABLE_READ_OLD_API);
-        DorisStreamOptions options = new DorisStreamOptions(properties);
-
-        List<String> actual = new ArrayList<>();
-        try (CloseableIterator<List<?>> iterator =
-                env.addSource(
-                                new DorisSourceFunction(
-                                        options, new SimpleListDeserializationSchema()))
-                        .executeAndCollect()) {
-            while (iterator.hasNext()) {
-                actual.add(iterator.next().toString());
-            }
-        }
-        List<String> expected = Arrays.asList("[doris, 18]", "[flink, 10]", "[apache, 12]");
-        checkResultInAnyOrder("testOldSourceApi", expected.toArray(), actual.toArray());
-    }
-
     @Test
     public void testSource() throws Exception {
         initializeTable(TABLE_READ, DataModel.AGGREGATE);
@@ -170,6 +144,32 @@ public void testSource() throws Exception {
         checkResultInAnyOrder("testSource", expected.toArray(), actual.toArray());
     }
 
+    @Test
+    public void testOldSourceApi() throws Exception {
+        initializeTable(TABLE_READ_OLD_API, DataModel.UNIQUE);
+        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+        env.setParallelism(DEFAULT_PARALLELISM);
+        Properties properties = new Properties();
+        properties.put("fenodes", getFenodes());
+        properties.put("username", getDorisUsername());
+        properties.put("password", getDorisPassword());
+        properties.put("table.identifier", DATABASE + "." + TABLE_READ_OLD_API);
+        DorisStreamOptions options = new DorisStreamOptions(properties);
+
+        List<String> actual = new ArrayList<>();
+        try (CloseableIterator<List<?>> iterator =
+                env.addSource(
+                                new DorisSourceFunction(
+                                        options, new SimpleListDeserializationSchema()))
+                        .executeAndCollect()) {
+            while (iterator.hasNext()) {
+                actual.add(iterator.next().toString());
+            }
+        }
+        List<String> expected = Arrays.asList("[doris, 18]", "[flink, 10]", "[apache, 12]");
+        checkResultInAnyOrder("testOldSourceApi", expected.toArray(), actual.toArray());
+    }
+
     @Test
     public void testTableSource() throws Exception {
         initializeTable(TABLE_READ_TBL, DataModel.DUPLICATE);

Original file line number	Diff line number	Diff line change
`@@ -47,8 +47,8 @@`
`47`	`47`	`*`
`48`	`48`	`* <p>Core batch stream load writer shared across Flink versions.`
`49`	`49`	`*`
`50`		`- * <p>Version-specific modules should wrap this class (for example via DorisBatchWriterV1 /`
`51`		`- * DorisBatchWriterV2) and implement the appropriate Flink sink interfaces there.`
	`50`	`+ * <p>Version-specific modules should wrap this class (for example via DorisBatchWriterAdapter and`
	`51`	`+ * implement the appropriate Flink sink interfaces there.`
`52`	`52`	`*/`
`53`	`53`	`public class DorisBatchWriter<IN> {`
`54`	`54`	`private static final Logger LOG = LoggerFactory.getLogger(DorisBatchWriter.class);`
Original file line number	Diff line number	Diff line change
`@@ -46,7 +46,7 @@ public class DorisSinkTest {`
`46`	`46`	`private MockedStatic<BackendUtil> backendUtilMockedStatic;`
`47`	`47`
`48`	`48`	`@Before`
`49`		`- public void setUp() throws Exception{`
	`49`	`+ public void setUp() throws Exception {`
`50`	`50`	`backendUtilMockedStatic = mockStatic(BackendUtil.class);`
`51`	`51`	`backendUtilMockedStatic.when(() -> BackendUtil.tryHttpConnection(any())).thenReturn(true);`
`52`	`52`	`}`