This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12568-6633b0b461c3b038883d04cee1469f15d62d7d73 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
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);
