This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new e5d8f94414 [Fix][API] Honor multi-table failure policy during timer
flush (#11537)
e5d8f94414 is described below
commit e5d8f94414e8e4a504615cd44511aea9f49c78f6
Author: Doyeon Kim <[email protected]>
AuthorDate: Sat Jul 25 21:48:29 2026 +0900
[Fix][API] Honor multi-table failure policy during timer flush (#11537)
---
.../common/multitable/MultiTableFailurePhase.java | 1 +
.../sink/multitablesink/MultiTableSinkWriter.java | 16 +++-
.../multitablesink/MultiTableSinkWriterTest.java | 90 ++++++++++++++++++++++
3 files changed, 106 insertions(+), 1 deletion(-)
diff --git
a/seatunnel-api/src/main/java/org/apache/seatunnel/api/common/multitable/MultiTableFailurePhase.java
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/common/multitable/MultiTableFailurePhase.java
index 3fafecef94..f2d13a7b78 100644
---
a/seatunnel-api/src/main/java/org/apache/seatunnel/api/common/multitable/MultiTableFailurePhase.java
+++
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/common/multitable/MultiTableFailurePhase.java
@@ -24,6 +24,7 @@ public enum MultiTableFailurePhase {
SINK_INIT("sink_init"),
SAVE_MODE("save_mode"),
RUNTIME_WRITE("runtime_write"),
+ TIMER_FLUSH("timer_flush"),
CHECKPOINT("checkpoint"),
CLOSE("close");
diff --git
a/seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/multitablesink/MultiTableSinkWriter.java
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/multitablesink/MultiTableSinkWriter.java
index 5734566709..39bc951fbc 100644
---
a/seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/multitablesink/MultiTableSinkWriter.java
+++
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/multitablesink/MultiTableSinkWriter.java
@@ -725,7 +725,21 @@ public class MultiTableSinkWriter
for (SinkIdentifier id : sinkWritersWithIndex.get(i).keySet())
{
SinkContextProxy proxy = proxyContexts.get(id);
if (proxy != null && proxy.getFlushAction() != null) {
- proxy.getFlushAction().run();
+ try {
+ proxy.getFlushAction().run();
+ } catch (InterruptedException error) {
+ Thread.currentThread().interrupt();
+ throw error;
+ } catch (Exception error) {
+ if (failurePolicy.continueOtherTables()) {
+ handleTableFailure(
+ id.getTableIdentifier(),
+ MultiTableFailurePhase.TIMER_FLUSH,
+ error);
+ continue;
+ }
+ throw error;
+ }
}
}
}
diff --git
a/seatunnel-api/src/test/java/org/apache/seatunnel/api/sink/multitablesink/MultiTableSinkWriterTest.java
b/seatunnel-api/src/test/java/org/apache/seatunnel/api/sink/multitablesink/MultiTableSinkWriterTest.java
index c068921f0f..17a87d3d16 100644
---
a/seatunnel-api/src/test/java/org/apache/seatunnel/api/sink/multitablesink/MultiTableSinkWriterTest.java
+++
b/seatunnel-api/src/test/java/org/apache/seatunnel/api/sink/multitablesink/MultiTableSinkWriterTest.java
@@ -345,6 +345,96 @@ public class MultiTableSinkWriterTest {
Assertions.assertDoesNotThrow(multiTableSinkWriter::close);
}
+ @Test
+ public void testAggregatedFlushIsolatesFailedTableAndFlushesHealthyTable()
{
+ Map<SinkIdentifier, SinkWriter<SeaTunnelRow, ?, ?>> sinkWriters = new
HashMap<>();
+ Map<SinkIdentifier, SinkWriter.Context> sinkWritersContext = new
HashMap<>();
+ Map<SinkIdentifier, SinkContextProxy> proxyContexts = new HashMap<>();
+ SinkIdentifier failedIdentifier = SinkIdentifier.of("test.failed", 0);
+ SinkIdentifier healthyIdentifier = SinkIdentifier.of("test.healthy",
1);
+ sinkWriters.put(failedIdentifier, new TestSinkWriter());
+ sinkWriters.put(healthyIdentifier, new TestSinkWriter());
+ sinkWritersContext.put(failedIdentifier, new TestSinkWriterContext());
+ sinkWritersContext.put(healthyIdentifier, new TestSinkWriterContext());
+
+ AtomicInteger failedFlushCount = new AtomicInteger();
+ SinkContextProxy failedProxy = new SinkContextProxy(0, 2, new
TestSinkWriterContext());
+ failedProxy.registerFlushAction(
+ () -> {
+ failedFlushCount.incrementAndGet();
+ throw new IOException("intentional flush failure");
+ });
+ AtomicInteger healthyFlushCount = new AtomicInteger();
+ SinkContextProxy healthyProxy = new SinkContextProxy(1, 2, new
TestSinkWriterContext());
+ healthyProxy.registerFlushAction(healthyFlushCount::incrementAndGet);
+ proxyContexts.put(failedIdentifier, failedProxy);
+ proxyContexts.put(healthyIdentifier, healthyProxy);
+
+ MultiTableSinkWriter multiTableSinkWriter =
+ new MultiTableSinkWriter(
+ sinkWriters,
+ 2,
+ sinkWritersContext,
+ MultiTableFailurePolicy.CONTINUE_OTHER_TABLES,
+ JobMode.BATCH);
+
+ Assertions.assertDoesNotThrow(() ->
multiTableSinkWriter.aggregatedFlush(proxyContexts));
+ Assertions.assertEquals(1, failedFlushCount.get());
+ Assertions.assertEquals(1, healthyFlushCount.get());
+
+ Assertions.assertDoesNotThrow(() ->
multiTableSinkWriter.aggregatedFlush(proxyContexts));
+ Assertions.assertEquals(1, failedFlushCount.get());
+ Assertions.assertEquals(2, healthyFlushCount.get());
+
+ IOException closeException =
+ Assertions.assertThrows(IOException.class,
multiTableSinkWriter::close);
+
Assertions.assertTrue(closeException.getMessage().contains("test.failed"));
+
Assertions.assertTrue(closeException.getMessage().contains("phase=timer_flush"));
+ }
+
+ @Test
+ public void testAggregatedFlushFailFastStopsAtFirstFailure() {
+ Map<SinkIdentifier, SinkWriter<SeaTunnelRow, ?, ?>> sinkWriters = new
HashMap<>();
+ Map<SinkIdentifier, SinkWriter.Context> sinkWritersContext = new
HashMap<>();
+ Map<SinkIdentifier, SinkContextProxy> proxyContexts = new HashMap<>();
+ SinkIdentifier failedIdentifier = SinkIdentifier.of("test.failed", 0);
+ SinkIdentifier healthyIdentifier = SinkIdentifier.of("test.healthy",
1);
+ sinkWriters.put(failedIdentifier, new TestSinkWriter());
+ sinkWriters.put(healthyIdentifier, new TestSinkWriter());
+ sinkWritersContext.put(failedIdentifier, new TestSinkWriterContext());
+ sinkWritersContext.put(healthyIdentifier, new TestSinkWriterContext());
+
+ AtomicInteger failedFlushCount = new AtomicInteger();
+ SinkContextProxy failedProxy = new SinkContextProxy(0, 2, new
TestSinkWriterContext());
+ failedProxy.registerFlushAction(
+ () -> {
+ failedFlushCount.incrementAndGet();
+ throw new IOException("intentional flush failure");
+ });
+ AtomicInteger healthyFlushCount = new AtomicInteger();
+ SinkContextProxy healthyProxy = new SinkContextProxy(1, 2, new
TestSinkWriterContext());
+ healthyProxy.registerFlushAction(healthyFlushCount::incrementAndGet);
+ proxyContexts.put(failedIdentifier, failedProxy);
+ proxyContexts.put(healthyIdentifier, healthyProxy);
+
+ MultiTableSinkWriter multiTableSinkWriter =
+ new MultiTableSinkWriter(
+ sinkWriters,
+ 2,
+ sinkWritersContext,
+ MultiTableFailurePolicy.FAIL_FAST,
+ JobMode.BATCH);
+
+ IOException flushException =
+ Assertions.assertThrows(
+ IOException.class,
+ () ->
multiTableSinkWriter.aggregatedFlush(proxyContexts));
+ Assertions.assertEquals("intentional flush failure",
flushException.getMessage());
+ Assertions.assertEquals(1, failedFlushCount.get());
+ Assertions.assertEquals(0, healthyFlushCount.get());
+ Assertions.assertDoesNotThrow(multiTableSinkWriter::close);
+ }
+
@Test
public void testIsolatedFailureMarkerRecognition() {
String message =
