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 fa001a4e47 [Fix][Connector-V2] Remove fixed sleep in AssertSinkWriter
close and assert own table only (#11995)
fa001a4e47 is described below
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);
+ }
+}
