[Chore] just test infinite loop case for doris connector

seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/sink/writer/RecordBuffer.java

@@ -85,6 +85,7 @@ public void stopBufferData() throws IOException {
             if (!isEmpty) {
                 ByteBuffer byteBuffer = null;
                 while (byteBuffer == null) {
+                    log.info("stopping buffer data...");
                     checkErrorMessageByStreamLoad();
                     byteBuffer = writeQueue.poll(100, TimeUnit.MILLISECONDS);
                 }

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/java/org/apache/seatunnel/e2e/connector/doris/DorisErrorIT.java

@@ -25,6 +25,7 @@
 import org.apache.seatunnel.e2e.common.junit.TestContainerExtension;
 
 import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.TestTemplate;
 import org.testcontainers.containers.Container;
 
@@ -57,6 +58,7 @@ public class DorisErrorIT extends AbstractDorisIT {
                 Assertions.assertEquals(0, extraCommands.getExitCode(), extraCommands.getStderr());
             };
 
+    @Disabled
     @TestTemplate
     @DisabledOnContainer(
             value = {},

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/java/org/apache/seatunnel/e2e/connector/doris/DorisIT.java

@@ -60,6 +60,7 @@
 @Slf4j
 public class DorisIT extends AbstractDorisIT {
     private static final String UNIQUE_TABLE = "doris_e2e_unique_table";
+    private static final String CAST_ERROR_SINK_TABLE = "doris_e2e_cast_error_sink_table";
     private static final String DUPLICATE_TABLE = "doris_duplicate_table";
     private static final String sourceDB = "e2e_source";
     private static final String sinkDB = "e2e_sink";
@@ -170,6 +171,22 @@ public class DorisIT extends AbstractDorisIT {
                 Assertions.assertEquals(0, extraCommands.getExitCode(), extraCommands.getStderr());
             };
 
+    @TestTemplate
+    public void testDorisCastError(TestContainer container)
+            throws IOException, InterruptedException {
+        initializeJdbcTable();
+        batchInsertUniqueTableData();
+        // Test that the task can terminate normally instead of being blocked when Doris parsing
+        // error occurs
+        // Verify that when source and target table data types are incompatible (e.g., varchar to
+        // bitmap),
+        // the task should fail gracefully and exit rather than hang indefinitely
+        Container.ExecResult execResult =
+                container.executeJob("/doris_source_and_sink_with_cast_error.conf");
+        Assertions.assertEquals(1, execResult.getExitCode());
+        Assertions.assertTrue(execResult.getStderr().contains("can not cast from origin type"));
+    }
+
     @TestTemplate
     public void testCustomSql(TestContainer container) throws IOException, InterruptedException {
         initializeJdbcTable();
@@ -422,6 +439,8 @@ protected void initializeJdbcTable() {
                 // create source and sink table
                 statement.execute(createUniqueTableForTest(sourceDB));
                 statement.execute(createDuplicateTableForTest(sourceDB));
+
+                statement.execute(createTypeCastErrorSinkTableForTest(sinkDB));
                 log.info("create source and sink table succeed");
             } catch (SQLException e) {
                 throw new RuntimeException("Initializing table failed!", e);
@@ -477,6 +496,52 @@ private String createUniqueTableForTest(String db) {
         return String.format(createTableSql, db, UNIQUE_TABLE);
     }
 
+    private String createTypeCastErrorSinkTableForTest(String db) {
+        // The type of column MAP_VARCHAR_STRING in sink table bitmap, source table is varchar type.
+        // In this case, doris will report an error. After seatunnel receives the error msg from
+        // doris,
+        // it should stop the task normally instead of being blocked and unable to terminate.
+        String createTableSql =
+                "create table if not exists `%s`.`%s`(\n"
+                        + "F_ID bigint null,\n"
+                        + "F_INT int null,\n"
+                        + "F_BIGINT bigint null,\n"
+                        + "F_TINYINT tinyint null,\n"
+                        + "F_SMALLINT smallint null,\n"
+                        + "F_DECIMAL decimal(18,6) null,\n"
+                        + "F_LARGEINT largeint null,\n"
+                        + "F_BOOLEAN boolean null,\n"
+                        + "F_DOUBLE double null,\n"
+                        + "F_FLOAT float null,\n"
+                        + "F_CHAR char null,\n"
+                        + "F_VARCHAR_11 ARRAY<int>,\n"
+                        + "F_STRING string null,\n"
+                        + "F_DATETIME_P datetime(6),\n"
+                        + "F_DATETIME datetime,\n"
+                        + "F_DATE date,\n"
+                        + "MAP_VARCHAR_BOOLEAN map<varchar(200),boolean>,\n"
+                        + "MAP_CHAR_TINYINT MAP<CHAR, TINYINT>,\n"
+                        + "MAP_STRING_SMALLINT MAP<STRING, SMALLINT>,\n"
+                        + "MAP_INT_INT MAP<INT, INT>,\n"
+                        + "MAP_TINYINT_BIGINT MAP<TINYINT, BIGINT>,\n"
+                        + "MAP_SMALLINT_LARGEINT MAP<SMALLINT, LARGEINT>,\n"
+                        + "MAP_BIGINT_FLOAT MAP<BIGINT, FLOAT>,\n"
+                        + "MAP_LARGEINT_DOUBLE MAP<LARGEINT, DOUBLE>,\n"
+                        + "MAP_STRING_DECIMAL MAP<STRING, DECIMAL(10,2)>,\n"
+                        + "MAP_DECIMAL_DATE MAP<DECIMAL(10,2), DATE>,\n"
+                        + "MAP_DATE_DATETIME MAP<DATE, DATETIME>,\n"
+                        + "MAP_DATETIME_CHAR MAP<DATETIME, CHAR(20)>,\n"
+                        + "MAP_CHAR_VARCHAR MAP<CHAR(20), VARCHAR(255)>,\n"
+                        + "MAP_VARCHAR_STRING bitmap NOT NULL\n"
+                        + ")\n"
+                        + "UNIQUE KEY(`F_ID`)\n"
+                        + "DISTRIBUTED BY HASH(`F_ID`) BUCKETS 1\n"
+                        + "properties(\n"
+                        + "\"replication_allocation\" = \"tag.location.default: 1\""
+                        + ");";
+        return String.format(createTableSql, db, CAST_ERROR_SINK_TABLE);
+    }
+
     private String createDuplicateTableForTest(String db) {
         String createDuplicateTableSql =
                 "create table if not exists `%s`.`%s`(\n"

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/java/org/apache/seatunnel/e2e/connector/doris/DorisMultiReadIT.java

@@ -27,6 +27,7 @@
 
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.TestTemplate;
 import org.testcontainers.containers.Container;
 
@@ -117,6 +118,7 @@ public class DorisMultiReadIT extends AbstractDorisIT {
                 Assertions.assertEquals(0, extraCommands.getExitCode(), extraCommands.getStderr());
             };
 
+    @Disabled
     @TestTemplate
     public void testDorisMultiRead(TestContainer container)
             throws IOException, InterruptedException {

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/resources/doris_source_and_sink_with_cast_error.conf

@@ -0,0 +1,53 @@
+#
+# 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.
+#
+
+env{
+  parallelism = 1
+  job.mode = "BATCH"
+}
+
+source{
+  Doris {
+      fenodes = "doris_e2e:8030"
+      username = root
+      password = ""
+      database = "e2e_source"
+      table = "doris_e2e_unique_table"
+      doris.batch.size = 10
+  }
+}
+
+transform {}
+
+sink{
+  Doris {
+          fenodes = "doris_e2e:8030"
+          username = root
+          password = ""
+          table.identifier = "e2e_sink.doris_e2e_cast_error_sink_table"
+          sink.enable-2pc = "false"
+          sink.label-prefix = "test_json"
+          sink.check-interval = 20000
+          doris.batch.size = 10
+          sink.buffer-count = 2
+          sink.buffer-size = 5120
+          doris.config = {
+              format="json"
+              read_json_by_line="true"
+          }
+      }
+  }