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-11995-c6b363af5c18702102dcc5536f6adf47b0ccb067 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit fa001a4e474841c35869d821584dbb0fbc78ff6c Author: corgy-w <[email protected]> AuthorDate: Tue Sep 1 15:40:21 2026 +0000 [Fix][Connector-V2] Remove fixed sleep in AssertSinkWriter close and assert own table only (#11995) --- .../seatunnel/assertion/sink/AssertSinkWriter.java | 15 ++-- .../assertion/sink/AssertSinkWriterCloseTest.java | 87 ++++++++++++++++++++++ 2 files changed, 93 insertions(+), 9 deletions(-) diff --git a/seatunnel-connectors-v2/connector-assert/src/main/java/org/apache/seatunnel/connectors/seatunnel/assertion/sink/AssertSinkWriter.java b/seatunnel-connectors-v2/connector-assert/src/main/java/org/apache/seatunnel/connectors/seatunnel/assertion/sink/AssertSinkWriter.java index 1f4b1d017f..0cb4a18adb 100644 --- a/seatunnel-connectors-v2/connector-assert/src/main/java/org/apache/seatunnel/connectors/seatunnel/assertion/sink/AssertSinkWriter.java +++ b/seatunnel-connectors-v2/connector-assert/src/main/java/org/apache/seatunnel/connectors/seatunnel/assertion/sink/AssertSinkWriter.java @@ -49,7 +49,6 @@ public class AssertSinkWriter extends AbstractSinkWriter<SeaTunnelRow, Void> private static final Map<String, LongAccumulator> LONG_ACCUMULATOR = new ConcurrentHashMap<>(); private static final Set<String> TABLE_NAMES = new CopyOnWriteArraySet<>(); private final String catalogTableName; - private final long WAIT_SINK_WRITER_COMPLETE_TIME = 1000L; public AssertSinkWriter( SeaTunnelRowType seaTunnelRowType, @@ -103,16 +102,14 @@ public class AssertSinkWriter extends AbstractSinkWriter<SeaTunnelRow, Void> @Override public void close() { - try { - // When there are multiple AssertSinkWriters, some Sinks will run first, so let it wait - // for other Sinks, otherwise it will make incorrect judgments - Thread.sleep(WAIT_SINK_WRITER_COMPLETE_TIME); - } catch (InterruptedException e) { - throw new RuntimeException(e); - } if (!assertRowRules.isEmpty()) { assertRowRules.entrySet().stream() - .filter(entry -> !entry.getValue().isEmpty()) + .filter( + entry -> + !entry.getValue().isEmpty() + && (assertRowRules.size() == 1 + || entry.getKey() + .equals(this.catalogTableName))) .forEach( entry -> { List<AssertFieldRule.AssertRule> assertRules = entry.getValue(); diff --git a/seatunnel-connectors-v2/connector-assert/src/test/java/org/apache/seatunnel/connectors/seatunnel/assertion/sink/AssertSinkWriterCloseTest.java b/seatunnel-connectors-v2/connector-assert/src/test/java/org/apache/seatunnel/connectors/seatunnel/assertion/sink/AssertSinkWriterCloseTest.java new file mode 100644 index 0000000000..75704dc410 --- /dev/null +++ b/seatunnel-connectors-v2/connector-assert/src/test/java/org/apache/seatunnel/connectors/seatunnel/assertion/sink/AssertSinkWriterCloseTest.java @@ -0,0 +1,87 @@ +/* + * 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. + */ + +package org.apache.seatunnel.connectors.seatunnel.assertion.sink; + +import org.apache.seatunnel.api.table.type.BasicType; +import org.apache.seatunnel.api.table.type.SeaTunnelRow; +import org.apache.seatunnel.api.table.type.SeaTunnelRowType; +import org.apache.seatunnel.connectors.seatunnel.assertion.exception.AssertConnectorException; +import org.apache.seatunnel.connectors.seatunnel.assertion.rule.AssertFieldRule; +import org.apache.seatunnel.connectors.seatunnel.assertion.rule.AssertTableRule; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +public class AssertSinkWriterCloseTest { + + private static final SeaTunnelRowType ROW_TYPE = + new SeaTunnelRowType(new String[] {"id"}, new BasicType[] {BasicType.INT_TYPE}); + + private static AssertFieldRule.AssertRule minRowRule(int minRows) { + AssertFieldRule.AssertRule rule = new AssertFieldRule.AssertRule(); + rule.setRuleType(AssertFieldRule.AssertRuleType.MIN_ROW); + rule.setRuleValue((double) minRows); + return rule; + } + + @Test + public void testCloseOnlyAssertsOwnTableWhenMultipleTables() { + // Writer A receives one row; table B never receives any row. The old implementation + // slept 1s and asserted every table's rules in close(), so B's MIN_ROW=1 (0 rows + // written) made the writer for table A fail. The fixed close() only asserts the + // table this writer is responsible for, so it must succeed without depending on + // other writers' progress. + Map<String, List<AssertFieldRule.AssertRule>> assertRowRules = new HashMap<>(); + assertRowRules.put("tableA_mult", Collections.singletonList(minRowRule(1))); + assertRowRules.put("tableB_other", Collections.singletonList(minRowRule(1))); + + AssertSinkWriter writerA = + new AssertSinkWriter( + ROW_TYPE, + Collections.emptyMap(), + assertRowRules, + new AssertTableRule(Collections.emptyList()), + "tableA_mult"); + writerA.write(new SeaTunnelRow(new Object[] {1})); + Assertions.assertDoesNotThrow(writerA::close); + } + + @Test + public void testCloseThrowsWhenOwnTableRuleNotMet() { + // MIN_ROW=2 for the writer's own table but only 1 row was written: close() must + // still validate the writer's own table and fail. + Map<String, List<AssertFieldRule.AssertRule>> assertRowRules = new HashMap<>(); + assertRowRules.put("tableA_fail", Collections.singletonList(minRowRule(2))); + assertRowRules.put("tableB_other", Collections.singletonList(minRowRule(0))); + + AssertSinkWriter writerA = + new AssertSinkWriter( + ROW_TYPE, + Collections.emptyMap(), + assertRowRules, + new AssertTableRule(Collections.emptyList()), + "tableA_fail"); + writerA.write(new SeaTunnelRow(new Object[] {1})); + Assertions.assertThrows(AssertConnectorException.class, writerA::close); + } +}
