[pull] dev from apache:dev

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java

@@ -455,18 +455,33 @@ public void testSourceKafkaJsonFormatErrorHandleWaySkipToConsole(TestContainer c
     @TestTemplate
     public void testSourceKafkaJsonFormatErrorHandleWayFailToConsole(TestContainer container)
             throws IOException, InterruptedException {
-        DefaultSeaTunnelRowSerializer serializer =
-                DefaultSeaTunnelRowSerializer.create(
-                        "test_topic_error_message",
-                        SEATUNNEL_ROW_TYPE,
-                        DEFAULT_FORMAT,
-                        DEFAULT_FIELD_DELIMITER,
-                        null);
-        generateTestData(serializer::serializeRow, 0, 100);
+        TextSerializationSchema serializer =
+                TextSerializationSchema.builder()
+                        .seaTunnelRowType(SEATUNNEL_ROW_TYPE)
+                        .delimiter(DEFAULT_FIELD_DELIMITER)
+                        .build();
+
+        generateTestData(
+                row -> {
+                    Object[] fields = row.getFields().clone();
+                    fields[0] = "bad_id_" + fields[0];
+                    SeaTunnelRow badRow = new SeaTunnelRow(fields);
+                    byte[] value = serializer.serialize(badRow);
+                    return new ProducerRecord<>("test_topic_error_message", null, value);
+                },
+                0,
+                100);
         Container.ExecResult execResult =
                 container.executeJob(
                         "/kafka/kafkasource_format_error_handle_way_fail_to_console.conf");
-        Assertions.assertEquals(1, execResult.getExitCode(), execResult.getStderr());
+        String serverLogs = container.getServerLogs();
+        Assertions.assertTrue(
+                execResult.getExitCode() != 0
+                        || serverLogs.contains("NumberFormatException")
+                        || serverLogs.contains("For input string"),
+                "Expected format error and job failure when format_error_handle_way = fail, "
+                        + "but exit code was "
+                        + execResult.getExitCode());
     }
 
     @TestTemplate

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/kafka_to_kafka_exactly_once_streaming.conf

@@ -41,4 +41,4 @@ sink{
         bootstrap.servers = "kafkaCluster:9092"
         semantics = EXACTLY_ONCE
     }
-}
+}

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/resources/fake-to-redis-test-readonly-hash.conf

@@ -18,6 +18,8 @@
 env {
   parallelism = 1
   job.mode = "BATCH"
+  # Disable restart strategy for this test - we expect immediate failure on readonly error
+  execution.restart.strategy = "no"
   shade.identifier = "base64"
 
   #spark config
@@ -64,4 +66,4 @@ sink {
     hash_key_field = "id"
     batch_size = 33
   }
-}
+}

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/resources/fake-to-redis-test-readonly-key.conf

@@ -18,6 +18,8 @@
 env {
   parallelism = 1
   job.mode = "BATCH"
+  # Disable restart strategy for this test - we expect immediate failure on readonly error
+  execution.restart.strategy = "no"
   shade.identifier = "base64"
 
   #spark config
@@ -63,4 +65,4 @@ sink {
     data_type = key
     batch_size = 33
   }
-}
+}

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/resources/fake-to-redis-test-readonly-list.conf

@@ -18,6 +18,8 @@
 env {
   parallelism = 1
   job.mode = "BATCH"
+  # Disable restart strategy for this test - we expect immediate failure on readonly error
+  execution.restart.strategy = "no"
   shade.identifier = "base64"
 
   #spark config
@@ -63,4 +65,4 @@ sink {
     data_type = list
     batch_size = 33
   }
-}
+}

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/resources/fake-to-redis-test-readonly-set.conf

@@ -18,6 +18,8 @@
 env {
   parallelism = 1
   job.mode = "BATCH"
+  # Disable restart strategy for this test - we expect immediate failure on readonly error
+  execution.restart.strategy = "no"
   shade.identifier = "base64"
 
   #spark config
@@ -63,4 +65,4 @@ sink {
     data_type = set
     batch_size = 33
   }
-}
+}

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/resources/fake-to-redis-test-readonly-zset.conf

@@ -18,6 +18,8 @@
 env {
   parallelism = 1
   job.mode = "BATCH"
+  # Disable restart strategy for this test - we expect immediate failure on readonly error
+  execution.restart.strategy = "no"
   shade.identifier = "base64"
 
   #spark config
@@ -63,4 +65,4 @@ sink {
     data_type = zset
     batch_size = 33
   }
-}
+}

seatunnel-e2e/seatunnel-e2e-common/src/test/java/org/apache/seatunnel/e2e/common/container/flink/AbstractTestFlinkContainer.java

@@ -58,7 +58,11 @@ public abstract class AbstractTestFlinkContainer extends AbstractTestContainer {
                     "jobmanager.rpc.address: jobmanager",
                     "taskmanager.numberOfTaskSlots: 10",
                     "parallelism.default: 4",
-                    "env.java.opts: -Doracle.jdbc.timezoneAsRegion=false");
+                    "env.java.opts: -Doracle.jdbc.timezoneAsRegion=false",
+                    // limit restart attempts in e2e to avoid infinite retries
+                    "restart-strategy: fixed-delay",
+                    "restart-strategy.fixed-delay.attempts: 2",
+                    "restart-strategy.fixed-delay.delay: 1000");
 
     protected static final String DEFAULT_DOCKER_IMAGE = "flink:1.13.6-scala_2.11";
 

seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-1/src/test/resources/data_validator_fail.conf

@@ -22,6 +22,8 @@
 env {
   job.mode = "BATCH"
   parallelism = 1
+  # Disable restart strategy for this test - we expect immediate failure on validation error
+  execution.restart.strategy = "no"
 }
 
 source {