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-12537-40d7fef09d203e2b4b1b8a3c5bdfd09ed0b87e58 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit a1084015eeb390a6c7f53d3ce7391217e427e4d9 Author: SEZ <[email protected]> AuthorDate: Thu Oct 1 15:06:53 2026 +0000 [Feature][seatunnel-cli] Parse job failures into structured fields (#12537) --- .../seatunnel_cli/diagnostics/__init__.py | 22 ++ seatunnel-cli/seatunnel_cli/diagnostics/errors.py | 305 +++++++++++++++++++++ seatunnel-cli/tests/test_diagnostics_errors.py | 195 +++++++++++++ 3 files changed, 522 insertions(+) diff --git a/seatunnel-cli/seatunnel_cli/diagnostics/__init__.py b/seatunnel-cli/seatunnel_cli/diagnostics/__init__.py new file mode 100644 index 0000000000..bb71562bd1 --- /dev/null +++ b/seatunnel-cli/seatunnel_cli/diagnostics/__init__.py @@ -0,0 +1,22 @@ +# +# 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. +# + +"""Job failure diagnosis helpers for the SeaTunnel AI CLI.""" + +from seatunnel_cli.diagnostics.errors import ParsedError, parse_error + +__all__ = ["ParsedError", "parse_error"] diff --git a/seatunnel-cli/seatunnel_cli/diagnostics/errors.py b/seatunnel-cli/seatunnel_cli/diagnostics/errors.py new file mode 100644 index 0000000000..6b25aeff08 --- /dev/null +++ b/seatunnel-cli/seatunnel_cli/diagnostics/errors.py @@ -0,0 +1,305 @@ +# +# 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. +# + +"""Turn a raw SeaTunnel failure into structured fields. + +The CLI currently feeds a truncated stack trace straight to the model, which +buries the two things that actually identify a failure -- the SeaTunnel error +code and the innermost ``Caused by`` -- under whatever happened to fit in the +character budget. This module extracts them with plain regexes so callers can +lead with the signal. + +Everything here is a pure function over a string: no I/O, no network, no model +calls. Unknown input yields ``category="unknown"`` and ``hint=None`` rather +than a guess -- a wrong hint costs more than a missing one, because it sends +the model (and the user) down the wrong path with false confidence. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass, field + +# org.apache.seatunnel.common.exception.SeaTunnelErrorCode#getErrorMessage +# formats every coded failure as: +# ErrorCode:[COMMON-22], ErrorDescription:[SeaTunnel write file '...' failed] +# The description itself may contain bracketed text (paths, SQL), so allow one +# level of nesting before closing. Deeper nesting falls back to the shortest +# match, which still yields a usable prefix. +# +# The namespace accepts mixed case on purpose: the prefix is whatever each +# connector hardcoded, and 13 of the 69 namespaces in the tree are PascalCase +# rather than SCREAMING_CASE (Doris, Hbase, GraphQL, AmazonSqs, SelectDB, +# TDengine, ...), so an upper-case-only class silently drops them. +_ERROR_CODE_RE = re.compile( + r"ErrorCode:\[(?P<ns>[A-Za-z][A-Za-z0-9_]*)-(?P<num>\d+)\],\s*" + r"ErrorDescription:\[(?P<desc>(?:[^\[\]]|\[[^\[\]]*\])*)\]" +) + +# "Caused by: org.apache.seatunnel.api.Foo: message" -- the last occurrence is +# the innermost cause, which is the one worth showing first. +_CAUSED_BY_RE = re.compile( + r"^[ \t]*Caused by:[ \t]*(?P<cls>[\w.$]+(?:Exception|Error|Throwable))" + r"(?::[ \t]*(?P<msg>.*))?$", + re.MULTILINE, +) + +# A bare "com.foo.BarException: message" line, used when there is no +# "Caused by" chain at all. +_EXCEPTION_LINE_RE = re.compile( + r"^[ \t]*(?P<cls>[\w.$]+(?:Exception|Error|Throwable))" + r"(?::[ \t]*(?P<msg>.*))?$", + re.MULTILINE, +) + +# Error-code namespaces that are not connectors. Everything else is treated as +# a connector namespace, so new connectors classify correctly without touching +# this module -- a hardcoded connector list would rot on every new connector. +_INFRASTRUCTURE_NAMESPACES = { + "API": "config", + "COMMON": "engine", + "SE": "engine", + "TRANSFORM_COMMON": "transform", + "JSONPATH_ERROR_CODE": "transform", +} + +# Failures that carry no SeaTunnel error code. These are JVM- and +# driver-level strings, matched case-insensitively and ordered most specific +# first: the first match wins, so a narrow pattern must precede a broad one. +_TEXT_PATTERNS: tuple[tuple[str, str, str, str], ...] = ( + ( + r"ClassNotFoundException|NoClassDefFoundError", + "missing_dependency", + "A class is missing from the classpath", + "Install the connector plugin and its JDBC driver into " + "$SEATUNNEL_HOME/plugins (or lib/) and restart the cluster.", + ), + ( + r"Access denied for user|password authentication failed" + r"|authentication failed|Login failed for user" + r"|SASL authentication failed", + "auth", + "The endpoint rejected the supplied credentials", + "Check the user/password options for this plugin, and that the " + "account may connect from the SeaTunnel host.", + ), + ( + r"UnknownHostException|Name or service not known" + r"|Temporary failure in name resolution", + "network", + "A hostname in the config did not resolve", + "Verify the host spelling; inside containers 'localhost' means the " + "container itself, not the host machine.", + ), + ( + r"Connection refused|No route to host|ConnectException", + "network", + "The endpoint refused the connection", + "Confirm the service is listening on that host and port and that no " + "firewall or network policy blocks it.", + ), + ( + r"OutOfMemoryError|GC overhead limit exceeded" + r"|unable to create new native thread", + "resource", + "The JVM ran out of memory or threads", + "Raise the worker heap, or lower parallelism / batch size so less " + "data is buffered at once.", + ), + ( + r"doesn't exist|does not exist|Unknown database|Unknown column" + r"|Invalid object name|Table or view not found", + "schema", + "A referenced table, column, or database is absent", + "Check the table and field names against the endpoint, including " + "case, and whether the sink needs save_mode to create them.", + ), + ( + r"FileNotFoundException|No such file or directory|NoSuchFileException", + "missing_path", + "A path in the config does not exist", + "Verify the path exists and is readable by the SeaTunnel process; " + "for cluster mode it must exist on every node.", + ), + ( + r"SocketTimeoutException|TimeoutException|Read timed out" + r"|timeout expired", + "timeout", + "An operation exceeded its time limit", + "Check endpoint health and latency, then raise the plugin's timeout " + "option if the endpoint is simply slow.", + ), + ( + r"Permission denied|AccessDeniedException|not authorized" + r"|AccessDenied", + "permission", + "The endpoint accepted the identity but refused the operation", + "Grant the account the required privileges on the target object " + "(for CDC sources this includes replication privileges).", + ), +) + +_COMPILED_TEXT_PATTERNS = tuple( + (re.compile(pattern, re.IGNORECASE), category, summary, hint) + for pattern, category, summary, hint in _TEXT_PATTERNS +) + +# Hints for coded failures, keyed by category rather than by individual code: +# the code set is large and changes, the remedy per class does not. +_CATEGORY_HINTS = { + "config": "Validate the plugin's options against its documented option " + "rule -- required options, and options that only apply to " + "certain formats.", + "transform": "Check the transform's field references and that the Zeta " + "SQL engine supports the expression used.", +} + +# Substrings dropped from a signature so the same failure on different rows, +# hosts, or attempts collapses to one key. +_VOLATILE_RE = re.compile(r"'[^']*'|\"[^\"]*\"|\b[0-9a-fA-F]{8,}\b|\d+") + + +@dataclass(frozen=True) +class ParsedError: + """Structured view of one failure. Absent fields stay ``None``.""" + + category: str = "unknown" + code: str | None = None + namespace: str | None = None + description: str | None = None + component: str | None = None + exception: str | None = None + root_cause: str | None = None + summary: str | None = None + hint: str | None = None + signature: str | None = None + codes: tuple[str, ...] = field(default_factory=tuple) + + def as_dict(self) -> dict: + """Drop empty fields so callers can render without None checks.""" + data = { + "category": self.category, + "code": self.code, + "namespace": self.namespace, + "description": self.description, + "component": self.component, + "exception": self.exception, + "root_cause": self.root_cause, + "summary": self.summary, + "hint": self.hint, + "signature": self.signature, + } + result = {k: v for k, v in data.items() if v} + if len(self.codes) > 1: + result["codes"] = list(self.codes) + return result + + def headline(self) -> str: + """One line naming the failure, for a compact report or tool result.""" + parts = [self.code or self.exception or "unknown failure"] + detail = self.description or self.root_cause or self.summary + if detail: + parts.append(detail) + return ": ".join(parts) + + +def _short_class(name: str) -> str: + return name.rsplit(".", 1)[-1] + + +def _signature(code: str | None, exception: str | None, text: str) -> str: + """Stable dedupe key: identity of the failure without its variable parts.""" + if code: + # An error code is already stable by construction, and its digits are + # exactly what tells one failure from another, so it must not go + # through _VOLATILE_RE -- that would turn JDBC-05 and JDBC-12 into the + # same "JDBC-?" key. + return code[:120] + if exception: + base = _short_class(exception) + else: + base = next((line.strip() for line in text.splitlines() if line.strip()), + "empty") + return _VOLATILE_RE.sub("?", base)[:120] + + +def parse_error(text: str | None) -> ParsedError: + """Extract error code, root cause, and a category from a raw failure.""" + # Anything that is not a string is treated as no input rather than raised + # on: this runs while the CLI is already reporting a failure, and a parser + # that throws there would replace the user's real error with its own. + if not isinstance(text, str) or not text.strip(): + return ParsedError(signature="empty") + + codes: list[str] = [] + first_desc: str | None = None + first_ns: str | None = None + for match in _ERROR_CODE_RE.finditer(text): + code = f"{match.group('ns')}-{match.group('num')}" + if code not in codes: + codes.append(code) + if first_desc is None: + first_ns = match.group("ns") + first_desc = " ".join(match.group("desc").split()) or None + + causes = _CAUSED_BY_RE.findall(text) + if causes: + # findall returns (cls, msg) pairs; the last is the innermost cause. + cause_cls, cause_msg = causes[-1] + exception = cause_cls + root_cause = " ".join(cause_msg.split()) or None + else: + exception, root_cause = None, None + bare = _EXCEPTION_LINE_RE.search(text) + if bare: + exception = bare.group("cls") + root_cause = " ".join((bare.group("msg") or "").split()) or None + + # Classify on the innermost cause when there is one: a generic + # SeaTunnelRuntimeException wrapper says less than what it wrapped. + category, summary, hint = "unknown", None, None + for pattern, cat, text_summary, text_hint in _COMPILED_TEXT_PATTERNS: + if pattern.search(text): + category, summary, hint = cat, text_summary, text_hint + break + + component = None + if first_ns: + if first_ns in _INFRASTRUCTURE_NAMESPACES: + coded_category = _INFRASTRUCTURE_NAMESPACES[first_ns] + else: + coded_category = "connector" + component = first_ns + # A concrete text match (auth, network, ...) explains the failure + # better than the namespace it surfaced in, so it takes precedence. + if category == "unknown": + category = coded_category + hint = _CATEGORY_HINTS.get(category) + + return ParsedError( + category=category, + code=codes[0] if codes else None, + namespace=first_ns, + description=first_desc, + component=component, + exception=exception, + root_cause=root_cause, + summary=summary, + hint=hint, + signature=_signature(codes[0] if codes else None, exception, text), + codes=tuple(codes), + ) diff --git a/seatunnel-cli/tests/test_diagnostics_errors.py b/seatunnel-cli/tests/test_diagnostics_errors.py new file mode 100644 index 0000000000..2513f6f4c6 --- /dev/null +++ b/seatunnel-cli/tests/test_diagnostics_errors.py @@ -0,0 +1,195 @@ +# +# 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. +# + +"""Tests for structured failure parsing.""" + +from seatunnel_cli.diagnostics import parse_error + + +# Shape taken from SeaTunnelErrorCode#getErrorMessage plus the wrapper chain +# Zeta puts around a connector failure. +JDBC_AUTH_TRACE = """ +org.apache.seatunnel.core.starter.exception.CommandExecuteException: SeaTunnel job executed failed +\tat org.apache.seatunnel.core.starter.seatunnel.command.ClientExecuteCommand.execute(ClientExecuteCommand.java:191) +Caused by: org.apache.seatunnel.common.exception.SeaTunnelRuntimeException: ErrorCode:[JDBC-05], ErrorDescription:[Connect to database failed] +\tat org.apache.seatunnel.connectors.seatunnel.jdbc.internal.JdbcOutputFormat.open(JdbcOutputFormat.java:142) +Caused by: java.sql.SQLException: Access denied for user 'bench'@'10.0.0.4' (using password: YES) +\tat com.mysql.cj.jdbc.exceptions.SQLError.createSQLException(SQLError.java:129) +""" + +MISSING_DRIVER_TRACE = """ +java.lang.RuntimeException: Failed to create sink writer +Caused by: java.lang.ClassNotFoundException: com.mysql.cj.jdbc.Driver +\tat java.base/java.net.URLClassLoader.findClass(URLClassLoader.java:445) +""" + +CONFIG_TRACE = ( + "org.apache.seatunnel.api.configuration.util.OptionValidationException: " + "ErrorCode:[API-02], ErrorDescription:[Option item validate failed]" +) + +NESTED_BRACKET_TRACE = ( + "ErrorCode:[COMMON-22], " + "ErrorDescription:[SeaTunnel write file [/tmp/out/part-0] failed]" +) + + +def test_extracts_code_root_cause_and_prefers_concrete_category(): + parsed = parse_error(JDBC_AUTH_TRACE) + assert parsed.code == "JDBC-05" + assert parsed.namespace == "JDBC" + assert parsed.component == "JDBC" + assert parsed.description == "Connect to database failed" + # The innermost cause, not the outer CommandExecuteException. + assert parsed.exception == "java.sql.SQLException" + assert "Access denied for user" in parsed.root_cause + # "auth" explains the failure; "connector" (from the JDBC namespace) does + # not, so the concrete text match must win. + assert parsed.category == "auth" + assert parsed.hint and "user/password" in parsed.hint + + +def test_headline_leads_with_code_and_description(): + assert parse_error(JDBC_AUTH_TRACE).headline() == ( + "JDBC-05: Connect to database failed" + ) + + +def test_classifies_missing_driver_without_an_error_code(): + parsed = parse_error(MISSING_DRIVER_TRACE) + assert parsed.code is None + assert parsed.category == "missing_dependency" + assert parsed.exception == "java.lang.ClassNotFoundException" + assert parsed.root_cause == "com.mysql.cj.jdbc.Driver" + assert parsed.hint and "plugins" in parsed.hint + + +def test_infrastructure_namespace_maps_to_config_not_connector(): + parsed = parse_error(CONFIG_TRACE) + assert parsed.code == "API-02" + assert parsed.category == "config" + # API is not a connector, so no component is claimed. + assert parsed.component is None + assert parsed.hint and "option rule" in parsed.hint + + +def test_mixed_case_namespace_is_recognized(): + # Most namespaces are SCREAMING_CASE, but 13 of the 69 in the tree are + # PascalCase. Both spellings are equally real, so both must parse. Codes + # and descriptions below are verbatim from DorisErrorCode/GraphQLErrorCode. + for text, code, namespace, description in ( + ( + "ErrorCode:[Doris-06], ErrorDescription:[backend client error]", + "Doris-06", + "Doris", + "backend client error", + ), + ( + "ErrorCode:[GraphQL-01], " + "ErrorDescription:[The operation of GraphQL is error]", + "GraphQL-01", + "GraphQL", + "The operation of GraphQL is error", + ), + ( + "ErrorCode:[JDBC-02], ErrorDescription:[No suitable driver found]", + "JDBC-02", + "JDBC", + "No suitable driver found", + ), + ): + parsed = parse_error(text) + assert parsed.code == code + assert parsed.namespace == namespace + assert parsed.description == description + # None of these are infrastructure namespaces, so each names itself. + assert parsed.component == namespace + assert parsed.category == "connector" + + +def test_description_may_contain_brackets(): + parsed = parse_error(NESTED_BRACKET_TRACE) + assert parsed.code == "COMMON-22" + assert parsed.description == "SeaTunnel write file [/tmp/out/part-0] failed" + + +def test_unknown_input_admits_it_rather_than_guessing(): + parsed = parse_error("something went sideways") + assert parsed.category == "unknown" + assert parsed.hint is None + assert parsed.code is None + + +def test_blank_input_is_safe(): + for blank in (None, "", " \n\t "): + parsed = parse_error(blank) + assert parsed.category == "unknown" + assert parsed.signature == "empty" + + +def test_non_string_input_does_not_raise(): + # This parser runs while the CLI is already reporting a failure. Throwing + # there would hide the user's real error behind a parser bug, so anything + # that is not text degrades to "unknown" instead. + for value in (123, b"ErrorCode:[JDBC-02]", ["trace"], {"errorMsg": "x"}): + parsed = parse_error(value) + assert parsed.category == "unknown" + assert parsed.signature == "empty" + assert parsed.code is None + + +def test_signature_collapses_volatile_detail(): + first = parse_error( + "Caused by: java.sql.SQLException: Access denied for user 'a'@'10.0.0.4'" + ) + second = parse_error( + "Caused by: java.sql.SQLException: Access denied for user 'b'@'10.0.0.9'" + ) + assert first.signature == second.signature + # A different failure must not collapse into the same key. + assert parse_error(MISSING_DRIVER_TRACE).signature != first.signature + + +def test_signature_keeps_the_error_code_number(): + # The digits in a code are what distinguish one failure from another, so + # two codes in the same namespace must not share a dedupe key. + driver = parse_error( + "ErrorCode:[JDBC-02], ErrorDescription:[No suitable driver found]" + ) + transaction = parse_error( + "ErrorCode:[JDBC-05], ErrorDescription:[transaction operation failed]" + ) + assert driver.signature == "JDBC-02" + assert transaction.signature == "JDBC-05" + assert driver.signature != transaction.signature + + +def test_all_codes_are_retained_in_order(): + parsed = parse_error( + "ErrorCode:[JDBC-05], ErrorDescription:[Connect to database failed]\n" + "Caused by: ErrorCode:[COMMON-01], ErrorDescription:[read file failed]" + ) + assert parsed.code == "JDBC-05" + assert parsed.codes == ("JDBC-05", "COMMON-01") + assert parsed.as_dict()["codes"] == ["JDBC-05", "COMMON-01"] + + +def test_as_dict_omits_empty_fields(): + data = parse_error("something went sideways").as_dict() + assert "hint" not in data + assert "code" not in data + assert data["category"] == "unknown"
