This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] 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 c0d2398c86 [Fix][API] Reject multi_table_sink_replica below 1 with a
diagnosable error (#12568)
c0d2398c86 is described below
commit c0d2398c86098062e7d6e40a847d97423301ea3c
Author: Sepuri Sai Krishna <[email protected]>
AuthorDate: Thu Oct 15 16:54:53 2026 +0000
[Fix][API] Reject multi_table_sink_replica below 1 with a diagnosable error
(#12568)
---
.../api/options/SinkConnectorCommonOptions.java | 3 +-
.../api/sink/multitablesink/MultiTableSink.java | 13 +++++
.../multitablesink/MultiTableSinkWriterTest.java | 67 ++++++++++++++++++++++
3 files changed, 82 insertions(+), 1 deletion(-)
diff --git
a/seatunnel-api/src/main/java/org/apache/seatunnel/api/options/SinkConnectorCommonOptions.java
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/options/SinkConnectorCommonOptions.java
index 1808aad506..22280fe6ee 100644
---
a/seatunnel-api/src/main/java/org/apache/seatunnel/api/options/SinkConnectorCommonOptions.java
+++
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/options/SinkConnectorCommonOptions.java
@@ -31,7 +31,8 @@ public class SinkConnectorCommonOptions extends
ConnectorCommonOptions {
Options.key("multi_table_sink_replica")
.intType()
.defaultValue(1)
- .withDescription("The replica number of multi table sink
writer");
+ .withDescription(
+ "The replica number of multi table sink writer.
Must be at least 1.");
@Experimental
public static Option<Map<String, String>> TABLE_OPTIONS =
diff --git
a/seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/multitablesink/MultiTableSink.java
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/multitablesink/MultiTableSink.java
index a3a1605857..34bf0dc07f 100644
---
a/seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/multitablesink/MultiTableSink.java
+++
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/multitablesink/MultiTableSink.java
@@ -90,6 +90,19 @@ public class MultiTableSink
this.sinks = context.getSinks();
this.replicaNum =
context.getOptions().get(SinkConnectorCommonOptions.MULTI_TABLE_SINK_REPLICA);
+ // replicaNum is the writer count per subtask and becomes the queue
count and the
+ // thread-pool size in MultiTableSinkWriter. Below 1 there is no
writer to route a row
+ // to, and Executors.newFixedThreadPool rejects the size with a
message-less
+ // IllegalArgumentException that names neither this option nor its
value. Fail here
+ // instead, with both.
+ if (replicaNum < 1) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Option '%s' must be at least 1, but was %d. Each
replica is one "
+ + "sink writer per subtask, so a value
below 1 leaves the "
+ + "multi-table sink with no writers.",
+
SinkConnectorCommonOptions.MULTI_TABLE_SINK_REPLICA.key(), replicaNum));
+ }
this.failurePolicy =
context.getOptions().get(MultiTableCommonOptions.MULTI_TABLE_FAILURE_POLICY);
this.tableRetryTimes =
context.getOptions().get(EnvCommonOptions.JOB_RETRY_TIMES);
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 9d9bcfed41..6035c52b4d 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
@@ -1021,6 +1021,73 @@ public class MultiTableSinkWriterTest {
Assertions.assertEquals(0, writersByIndex[2].getWriteCount());
}
+ @Test
+ public void testReplicaNumBelowOneIsRejectedWithTheOptionNamed() {
+ // multi_table_sink_replica has no lower bound at the option layer, so
0 and negative
+ // values reach MultiTableSink. replicaNum becomes the queue count and
the thread-pool
+ // size in MultiTableSinkWriter, where
Executors.newFixedThreadPool(queueSize * 2)
+ // rejects them with an IllegalArgumentException carrying no message
at all. The
+ // exception type is unchanged; what is asserted here is that it now
names the option
+ // and the offending value, and that it is raised before any writer is
created.
+ for (int replicaNum : new int[] {0, -1}) {
+ IllegalArgumentException exception =
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () ->
buildMultiTableSink(replicaNum));
+ Assertions.assertNotNull(exception.getMessage());
+ Assertions.assertTrue(
+ exception
+ .getMessage()
+
.contains(SinkConnectorCommonOptions.MULTI_TABLE_SINK_REPLICA.key()),
+ "message must name the option, but was: " +
exception.getMessage());
+ Assertions.assertTrue(
+
exception.getMessage().contains(String.valueOf(replicaNum)),
+ "message must name the rejected value, but was: " +
exception.getMessage());
+ }
+ }
+
+ @Test
+ public void testReplicaNumOfOneAndAboveStillBuilds() {
+ // Guards the lower bound against being set too high: the smallest
legal value and a
+ // larger one must both still construct, so the check above cannot
regress into
+ // rejecting valid configuration.
+ Assertions.assertDoesNotThrow(() -> buildMultiTableSink(1));
+ Assertions.assertDoesNotThrow(() -> buildMultiTableSink(4));
+ }
+
+ @Test
+ public void testReplicaNumDefaultsToOneWhenUnset() {
+ // The default must remain a legal value, otherwise the new check
would reject every
+ // job that does not set the option explicitly.
+ Map<String, Object> options = new HashMap<>();
+ options.put(EnvCommonOptions.JOB_RETRY_TIMES.key(), 0);
+ options.put(EnvCommonOptions.JOB_RETRY_INTERVAL_SECONDS.key(), 0);
+ Map<TablePath, SeaTunnelSink> sinks = new HashMap<>();
+ sinks.put(
+ TablePath.of("test.table"), new TestSeaTunnelSink(new
RecordingSinkWriter(false)));
+ Assertions.assertDoesNotThrow(
+ () ->
+ new MultiTableSink(
+ new MultiTableFactoryContext(
+ ReadonlyConfig.fromMap(options),
+
Thread.currentThread().getContextClassLoader(),
+ sinks)));
+ }
+
+ private MultiTableSink buildMultiTableSink(int replicaNum) {
+ Map<String, Object> options = new HashMap<>();
+ options.put(SinkConnectorCommonOptions.MULTI_TABLE_SINK_REPLICA.key(),
replicaNum);
+ options.put(EnvCommonOptions.JOB_RETRY_TIMES.key(), 0);
+ options.put(EnvCommonOptions.JOB_RETRY_INTERVAL_SECONDS.key(), 0);
+ Map<TablePath, SeaTunnelSink> sinks = new HashMap<>();
+ sinks.put(
+ TablePath.of("test.table"), new TestSeaTunnelSink(new
RecordingSinkWriter(false)));
+ return new MultiTableSink(
+ new MultiTableFactoryContext(
+ ReadonlyConfig.fromMap(options),
+ Thread.currentThread().getContextClassLoader(),
+ sinks));
+ }
+
private SeaTunnelRow buildRow(String tableId, int value) {
SeaTunnelRow row = new SeaTunnelRow(new Object[] {value});
row.setTableId(tableId);
