This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit b99f5b9955e7a57eb0ac3cf3fc9e0bd8714eca03 Author: Eunbin Son <[email protected]> AuthorDate: Sat Sep 19 23:02:53 2026 +0900 [e2e] Strip any trailing Spark error log from spark-sql output before comparison (#9971) --- .../apache/paimon/tests/SparkE2eOutputTest.java | 73 ++++++++++++++++++++++ .../java/org/apache/paimon/tests/SparkE2eTest.java | 36 +++++++---- 2 files changed, 97 insertions(+), 12 deletions(-) diff --git a/paimon-e2e-tests/src/test/java/org/apache/paimon/tests/SparkE2eOutputTest.java b/paimon-e2e-tests/src/test/java/org/apache/paimon/tests/SparkE2eOutputTest.java new file mode 100644 index 0000000000..6bf2fcf94f --- /dev/null +++ b/paimon-e2e-tests/src/test/java/org/apache/paimon/tests/SparkE2eOutputTest.java @@ -0,0 +1,73 @@ +/* + * 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.paimon.tests; + +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for {@link SparkE2eTest#stripTrailingSparkErrorLogs(String)}. Does not need Docker. */ +public class SparkE2eOutputTest { + + private static final String ROWS = "1\t10\tHi\n2\t20\tHello\n3\t30\tTable\n"; + + private static final String COARSE_GRAINED_SCHEDULER_ERROR = + "26/07/29 04:25:43 ERROR Utils: Uncaught exception in thread dispatcher-CoarseGrainedScheduler\n" + + "org.apache.spark.SparkException: Could not find CoarseGrainedScheduler.\n" + + "\tat org.apache.spark.rpc.netty.Dispatcher.postMessage(Dispatcher.scala:178)\n"; + + private static final String TRANSPORT_REQUEST_HANDLER_ERROR = + "26/08/30 07:28:25 ERROR TransportRequestHandler: Error while invoking RpcHandler#receive() for one-way message.\n" + + "org.apache.spark.SparkException: Could not find AppClient.\n" + + "\tat org.apache.spark.rpc.netty.Dispatcher.postMessage(Dispatcher.scala:178)\n" + + "\tat java.lang.Thread.run(Thread.java:748)\n"; + + @Test + public void testNoErrorLogIsUnchanged() { + assertThat(SparkE2eTest.stripTrailingSparkErrorLogs(ROWS)).isEqualTo(ROWS); + } + + @Test + public void testStripsCoarseGrainedSchedulerError() { + assertThat(SparkE2eTest.stripTrailingSparkErrorLogs(ROWS + COARSE_GRAINED_SCHEDULER_ERROR)) + .isEqualTo(ROWS); + } + + @Test + public void testStripsRepeatedTransportRequestHandlerErrors() { + assertThat( + SparkE2eTest.stripTrailingSparkErrorLogs( + ROWS + + TRANSPORT_REQUEST_HANDLER_ERROR + + TRANSPORT_REQUEST_HANDLER_ERROR)) + .isEqualTo(ROWS); + } + + @Test + public void testErrorOnFirstLineYieldsEmptyOutput() { + assertThat(SparkE2eTest.stripTrailingSparkErrorLogs(TRANSPORT_REQUEST_HANDLER_ERROR)) + .isEmpty(); + } + + @Test + public void testErrorWordInsideResultRowIsKept() { + String rows = "1\t10\tsome ERROR text\n2\t20\tERROR Foo\n"; + assertThat(SparkE2eTest.stripTrailingSparkErrorLogs(rows)).isEqualTo(rows); + } +} diff --git a/paimon-e2e-tests/src/test/java/org/apache/paimon/tests/SparkE2eTest.java b/paimon-e2e-tests/src/test/java/org/apache/paimon/tests/SparkE2eTest.java index e78b82e607..ad2e577041 100644 --- a/paimon-e2e-tests/src/test/java/org/apache/paimon/tests/SparkE2eTest.java +++ b/paimon-e2e-tests/src/test/java/org/apache/paimon/tests/SparkE2eTest.java @@ -18,6 +18,8 @@ package org.apache.paimon.tests; +import org.apache.paimon.annotation.VisibleForTesting; + import org.junit.jupiter.api.Test; import org.junit.jupiter.api.condition.DisabledOnJre; import org.slf4j.Logger; @@ -27,6 +29,8 @@ import org.testcontainers.containers.ContainerState; import java.util.Arrays; import java.util.UUID; +import java.util.regex.Matcher; +import java.util.regex.Pattern; import java.util.stream.Collectors; import static org.junit.jupiter.api.condition.JRE.JAVA_11; @@ -36,8 +40,14 @@ import static org.junit.jupiter.api.condition.JRE.JAVA_11; public class SparkE2eTest extends E2eReaderTestBase { private static final Logger LOG = LoggerFactory.getLogger(SparkE2eTest.class); - private static final String COARSE_GRAINED_SCHEDULER_SHUTDOWN_ERROR = - "ERROR Utils: Uncaught exception in thread dispatcher-CoarseGrainedScheduler"; + + /** + * Start of a Spark ERROR log line in the default log4j2 pattern ({@code %d{yy/MM/dd HH:mm:ss} + * %p %c{1}: %m%n%ex}), which spark-sql writes to stdout. Anchoring on the timestamp prefix + * avoids matching an "ERROR" word inside a result row. + */ + private static final Pattern SPARK_ERROR_LOG_LINE = + Pattern.compile("(?m)^\\d{2}/\\d{2}/\\d{2} \\d{2}:\\d{2}:\\d{2} ERROR "); public SparkE2eTest() { super(false, false, true); @@ -77,8 +87,7 @@ public class SparkE2eTest extends E2eReaderTestBase { LOG.info(execResult.getStderr()); throw new AssertionError("Failed when running spark sql."); } - String stdout = - stripCoarseGrainedSchedulerShutdownError(execResult.getStdout()); + String stdout = stripTrailingSparkErrorLogs(execResult.getStdout()); return Arrays.stream(stdout.split("\n")) .filter(s -> !s.contains("WARN")) .collect(Collectors.joining("\n")) @@ -86,14 +95,17 @@ public class SparkE2eTest extends E2eReaderTestBase { }); } - private static String stripCoarseGrainedSchedulerShutdownError(String stdout) { - int errorIndex = stdout.indexOf(COARSE_GRAINED_SCHEDULER_SHUTDOWN_ERROR); - if (errorIndex < 0) { - return stdout; - } - - int errorLineStart = stdout.lastIndexOf('\n', errorIndex); - return errorLineStart < 0 ? "" : stdout.substring(0, errorLineStart); + /** + * Drops everything from the first Spark ERROR log line onwards. When spark-sql exits, the + * driver shutdown races with RPC dispatch and may log after the query result, with exit code 0, + * e.g. {@code ERROR Utils: Uncaught exception in thread dispatcher-CoarseGrainedScheduler} or + * {@code ERROR TransportRequestHandler: Error while invoking RpcHandler#receive() for one-way + * message.} followed by a stack trace. + */ + @VisibleForTesting + static String stripTrailingSparkErrorLogs(String stdout) { + Matcher matcher = SPARK_ERROR_LOG_LINE.matcher(stdout); + return matcher.find() ? stdout.substring(0, matcher.start()) : stdout; } private ContainerState getSpark() {
