From a0f48af070d9860b74e70aedb8d6a5ae1fd07208 Mon Sep 17 00:00:00 2001 From: Jason Hunter Date: Tue, 1 Sep 2026 01:46:32 -0700 Subject: [PATCH 1/8] [bulk] Report driver-side failures politely too (closes #332) Workers have surfaced failures politely since #331; the driver had not. A denied connector read reached the user as ~320 lines: Glue's exception-analysis blob, Py4J's restatement, and a Java stack with AWS's actual sentence buried on line two, closing on "An error occurred while calling o304.load". Three parts. 1. root.py now routes every failure out of a verb through shared/driver_errors.py rather than re-raising. Understood -- a denial, expired credentials, throttling, or anything a verb already phrased as BulkExecutorError -- exits with one sentence. Unexpected prints its traceback once, for the user rather than for CloudWatch, and still exits with a one-line reason. Exiting rather than re-raising is the point: a re-raise is what summons Glue's blob and Py4J's duplicate. It catches BaseException, because a worker that calls exit() arrives as SystemExit through Py4J; driver_errors' own sys.exit is let through. 2. get_error_message now unwraps a Java/Py4J error. The SDK v1 pattern stopped matching when Glue moved to the v2 SDK (software.amazon.awssdk...DynamoDbException rather than com.amazonaws...AmazonDynamoDBException), so a denial's "message" was the whole stack. It takes the innermost cause, because the outer layer is usually Spark boilerplate ("Job aborted due to stage failure") while the cause is the sentence the user needs. 3. Per-site sentences where the net's generic one would be worse: sql's query errors, load's source read (naming the path and the format the user claimed) and its connector write, and the transform module named by --transform. The client suppresses Glue's blob on the driver banner as well as the worker one, and the drift guard now checks both constants against runner.py. --- tools/bulk_executor/client/src/runner.py | 9 +- .../src/python_modules/load/__init__.py | 11 +- .../python_modules/shared/driver_errors.py | 68 +++++++++++ .../src/python_modules/shared/errors.py | 24 +++- .../export/pipeline/transform_loader.py | 6 +- .../server/src/python_modules/sql.py | 3 +- tools/bulk_executor/server/src/root.py | 20 +++- .../export/pipeline/test_transform_loader.py | 4 +- .../tests/server/shared/test_driver_errors.py | 112 ++++++++++++++++++ .../shared/test_errors_message_extraction.py | 87 ++++++++++++++ tools/bulk_executor/tests/server/test_load.py | 8 +- tools/bulk_executor/tests/server/test_root.py | 60 ++++++++-- tools/bulk_executor/tests/server/test_sql.py | 15 ++- ...orker_failures_go_through_worker_errors.py | 18 +-- 14 files changed, 405 insertions(+), 40 deletions(-) create mode 100644 tools/bulk_executor/server/src/python_modules/shared/driver_errors.py create mode 100644 tools/bulk_executor/tests/server/shared/test_driver_errors.py create mode 100644 tools/bulk_executor/tests/server/shared/test_errors_message_extraction.py diff --git a/tools/bulk_executor/client/src/runner.py b/tools/bulk_executor/client/src/runner.py index 188b96fa..1c12ca04 100644 --- a/tools/bulk_executor/client/src/runner.py +++ b/tools/bulk_executor/client/src/runner.py @@ -55,12 +55,15 @@ ]) # Markers the job prints when it has already told the user what went wrong: a -# BulkExecutorError sentence, or a worker traceback that names the offending line. -# Either way Glue's exception analysis adds nothing but volume afterwards. -# The second must match worker_errors.UNEXPECTED_FAILURE_BANNER. +# BulkExecutorError sentence, or a traceback that names the offending line -- from a +# worker or from the driver. Either way Glue's exception analysis adds nothing but volume +# afterwards, and for a driver-side failure it also restates the same error through Py4J. +# The last two must match worker_errors.UNEXPECTED_FAILURE_BANNER and +# driver_errors.UNEXPECTED_FAILURE_BANNER (a guard test checks both). JOB_EXPLAINED_THE_FAILURE = ( 'BulkExecutorError', 'A worker failed in a way we did not expect. Traceback from the worker:', + 'The job failed in a way we did not expect. Traceback:', ) # Timing constants diff --git a/tools/bulk_executor/server/src/python_modules/load/__init__.py b/tools/bulk_executor/server/src/python_modules/load/__init__.py index fb9761ef..845bb87b 100644 --- a/tools/bulk_executor/server/src/python_modules/load/__init__.py +++ b/tools/bulk_executor/server/src/python_modules/load/__init__.py @@ -94,7 +94,12 @@ def run(job, spark_context, glue_context, parsed_args): dynamicFrame.printSchema() except Exception as e: - raise Exception(f"Failed to create DynamicFrame {e}") + # This is where Spark actually reads the source, so it fires on the everyday + # mistakes: --format json pointed at CSV, malformed JSON, an unreadable Parquet + # file. (A path that does not exist is caught earlier, before the job starts.) + raise BulkExecutorError( + f"Could not read the source at '{s3_path}' as {parsed_args.get('format')!r}: " + f"{get_error_message(e)}") from None if parsed_args.get('removeEmptyStringAttributes') is not None: log.debug(f"removeEmptyStringAttributes parameter was provided") @@ -114,7 +119,9 @@ def run(job, spark_context, glue_context, parsed_args): glue_context, df, table_name, parsed_args, write_rate=write_rate) log.info(f"Wrote {count} items to '{table_name}'") except Exception as e: - raise Exception(f"Error in writing to table: {get_error_message(e)}") from None + # The connector write: a read-only role or persistent throttling lands here, and + # get_error_message unwraps the Py4J stack to AWS's own sentence. + raise BulkExecutorError(f"Error in writing to table: {get_error_message(e)}") from None def check_s3_file_exists(s3_uri): """ diff --git a/tools/bulk_executor/server/src/python_modules/shared/driver_errors.py b/tools/bulk_executor/server/src/python_modules/shared/driver_errors.py new file mode 100644 index 00000000..a47f6a55 --- /dev/null +++ b/tools/bulk_executor/server/src/python_modules/shared/driver_errors.py @@ -0,0 +1,68 @@ +"""Surfacing a failure that happened on the driver. + +`worker_errors.py` handles the failures a worker records; this handles the ones the +driver hits directly -- a denied `.load()` through the Glue connector, a source file +that isn't the format the user claimed, a typo in `--transform`. The classification is +the same one, imported from there, because the question is the same: did we understand +this? + +- understood -- a denial, expired credentials, throttling, or something a verb already + phrased as a BulkExecutorError. One sentence, no traceback. +- unexpected -- anything else. The traceback is printed once, on the console, and the + job still exits with a one-line reason. + +Either way `root.py` exits rather than re-raising, so Glue's exception-analysis blob and +its Py4J restatement of the same error never reach the user. That matters most for the +connector: a denied read arrives as a Py4JJavaError whose str() is a 300-line Java stack +with AWS's sentence buried on line two, and re-raising it means the last thing the user +sees is `An error occurred while calling o304.load`. +""" + +import sys +import traceback + +from python_modules.shared.bulk_executor_error import BulkExecutorError +from python_modules.shared.logger import log +from python_modules.shared.worker_errors import classify_failure + +UNEXPECTED_FAILURE_BANNER = "The job failed in a way we did not expect. Traceback:" + +# Cap on what we hand to sys.exit(): Glue records it as the job's ErrorMessage and the +# client prints it as its closing line. AWS's authorization sentences run ~400 chars, +# which is worth keeping whole; a Java stack that slipped through is not. +MAX_REASON_CHARS = 800 + + +def _one_line(message): + """Collapse to a single line and bound the length. The reason is a closing line, not + a report -- anything longer has already been printed above it.""" + collapsed = " ".join(str(message).split()) + if len(collapsed) > MAX_REASON_CHARS: + collapsed = collapsed[:MAX_REASON_CHARS - 3] + "..." + return collapsed + + +def surface(exception): + """Report a driver-side failure and exit. Never returns. + + Called by root.py for anything that escapes a verb. + """ + understood, message = classify_failure(exception) + if not understood: + # Ours to debug, or a user's generator/transform: the frames are the report, and + # printing them ourselves means the client suppresses Glue's blob (it watches for + # this banner and for BulkExecutorError) and the closing line stays short. + print(UNEXPECTED_FAILURE_BANNER) + print(traceback.format_exc()) + # classify_failure already extracted the readable message; the type goes in front + # because an unexpected message alone can be as bare as "'pk'". + reason = f"{type(exception).__name__}: {message}" + else: + reason = message + + reason = _one_line(reason) + if isinstance(exception, BulkExecutorError): + log.error(f"BulkExecutorError: {reason}") + else: + log.error(reason) + sys.exit(reason) diff --git a/tools/bulk_executor/server/src/python_modules/shared/errors.py b/tools/bulk_executor/server/src/python_modules/shared/errors.py index 6f58fb2a..56598cba 100644 --- a/tools/bulk_executor/server/src/python_modules/shared/errors.py +++ b/tools/bulk_executor/server/src/python_modules/shared/errors.py @@ -47,7 +47,7 @@ def get_error_message(e): except json.JSONDecodeError: pass # fallback - # Look for DynamoDB exception message + # Look for DynamoDB exception message (AWS SDK v1, which Glue 4 and earlier used) dynamo_match = re.search( r'com\.amazonaws\.services\.dynamodbv2\.model\.AmazonDynamoDBException:\s*(.*?)\s*\(Service:', msg @@ -55,6 +55,28 @@ def get_error_message(e): if dynamo_match: return dynamo_match.group(1).strip() + # A Java exception relayed through Py4J. str() on one of these is the entire Java + # stack -- hundreds of frames -- with the sentence that matters on the second line: + # + # py4j.protocol.Py4JJavaError: An error occurred while calling o304.load. + # : software.amazon.awssdk.services.dynamodb.model.DynamoDbException: User: ... + # is not authorized to perform: dynamodb:Scan on resource: ... + # at software.amazon.awssdk...(DynamoDbException.java:113) + # + # Take the innermost cause's message: for a wrapped failure the outer layer is + # usually Spark boilerplate ("Job aborted due to stage failure") and the cause is + # the AWS sentence the user needs. The message can wrap onto continuation lines, so + # keep going until a stack frame or a new exception header. + java_causes = re.findall( + r'^(?:: |Caused by: )(?:[\w$]+\.)+([\w$]*(?:Exception|Error)): ' + r'(.*(?:\n(?!\s*(?:at |\.\.\. )|: |Caused by: ).*)*)', + msg, re.MULTILINE) + if java_causes: + _cls, detail = java_causes[-1] + detail = ' '.join(detail.split()) + if detail: + return detail + # ParseException handling if hasattr(e, 'desc'): # ParseException msg = e.desc diff --git a/tools/bulk_executor/server/src/python_modules/shared/export/pipeline/transform_loader.py b/tools/bulk_executor/server/src/python_modules/shared/export/pipeline/transform_loader.py index b5093885..9c033c0c 100644 --- a/tools/bulk_executor/server/src/python_modules/shared/export/pipeline/transform_loader.py +++ b/tools/bulk_executor/server/src/python_modules/shared/export/pipeline/transform_loader.py @@ -1,4 +1,5 @@ import importlib +from ...bulk_executor_error import BulkExecutorError def load_transform_module(module_name, transform_package): @@ -18,4 +19,7 @@ def load_transform_module(module_name, transform_package): try: return importlib.import_module(f"{transform_package}.{module_name}") except ImportError as e: - raise ImportError(f"Cannot import transform module '{module_name}' from '{transform_package}': {e}") + # --transform is user input, so a typo is a sentence rather than a traceback. + raise BulkExecutorError( + f"Cannot import transform module '{module_name}' from '{transform_package}': {e}" + ) from None diff --git a/tools/bulk_executor/server/src/python_modules/sql.py b/tools/bulk_executor/server/src/python_modules/sql.py index dbdf4c14..bb735134 100644 --- a/tools/bulk_executor/server/src/python_modules/sql.py +++ b/tools/bulk_executor/server/src/python_modules/sql.py @@ -92,7 +92,8 @@ def run(job, spark_context, glue_context, parsed_args): # without wrapping them into an opaque "SQL query error". raise except Exception as e: - raise Exception("SQL query error: " + get_error_message(e)) from None + # The query is the user's to fix, so this is a sentence, not a stack trace. + raise BulkExecutorError("SQL query error: " + get_error_message(e)) from None finally: # Ensure Spark session cleanup try: diff --git a/tools/bulk_executor/server/src/root.py b/tools/bulk_executor/server/src/root.py index 43d1c6c2..beb59902 100644 --- a/tools/bulk_executor/server/src/root.py +++ b/tools/bulk_executor/server/src/root.py @@ -7,6 +7,7 @@ from awsglue.job import Job from awsglue.transforms import * from pyspark.context import SparkContext +from python_modules.shared import driver_errors from python_modules.shared.bulk_executor_error import BulkExecutorError @@ -103,11 +104,20 @@ def _get_parsed_glue_job_args(argv): action_script_function = getattr(module, action_script_function_name) try: action_script_function(job, spark_context, glue_context, parsed_args) # Run the function - except BulkExecutorError as e: - log.error(f"BulkExecutorError: {e}") - sys.exit(str(e)) - except Exception as e: - raise # Just let it propagate + except BaseException as e: + # Everything a verb can fail with lands here, and shared/driver_errors.py + # decides what the user sees: a denial or other understood failure exits with + # one sentence, anything else prints its traceback first and then exits with a + # one-line reason. Exiting rather than re-raising is the point -- a re-raise + # hands the user Glue's exception-analysis blob plus Py4J's restatement of the + # same error, with AWS's actual sentence buried inside a Java stack (#332). + # + # BaseException, not Exception: a worker calling exit() reaches the driver as a + # SystemExit, and Py4J wraps KeyboardInterrupt-style aborts too. sys.exit from + # driver_errors.surface() raises SystemExit itself, so let that through. + if isinstance(e, SystemExit): + raise + driver_errors.surface(e) else: raise Exception(f"Could not find the function '{action_script_function_name}' within the module '{module_name}'.") diff --git a/tools/bulk_executor/tests/server/shared/export/pipeline/test_transform_loader.py b/tools/bulk_executor/tests/server/shared/export/pipeline/test_transform_loader.py index 363ae5ed..cad942e1 100644 --- a/tools/bulk_executor/tests/server/shared/export/pipeline/test_transform_loader.py +++ b/tools/bulk_executor/tests/server/shared/export/pipeline/test_transform_loader.py @@ -24,5 +24,7 @@ def test_default_incremental_is_callable(self): assert callable(module.transform_incremental_record) def test_load_nonexistent_module(self): - with pytest.raises(ImportError): + """--transform names the module, so a typo is a sentence, not a traceback (#332).""" + from python_modules.shared.bulk_executor_error import BulkExecutorError + with pytest.raises(BulkExecutorError, match="Cannot import transform module"): load_transform_module('nonexistent_module', 'python_modules.load_export.transform') diff --git a/tools/bulk_executor/tests/server/shared/test_driver_errors.py b/tools/bulk_executor/tests/server/shared/test_driver_errors.py new file mode 100644 index 00000000..09252dbe --- /dev/null +++ b/tools/bulk_executor/tests/server/shared/test_driver_errors.py @@ -0,0 +1,112 @@ +"""Unit tests for server/src/python_modules/shared/driver_errors.py. + +Covers surface(): an understood failure exits with one sentence and prints nothing else; +an unexpected one prints a banner plus its traceback and still exits with a one-line +reason; the reason is collapsed to a single line and bounded. + +The contract these protect is what #332 is about. A driver-side failure used to be +re-raised, which handed the user three copies of the same problem -- Glue's +exception-analysis blob, Py4J's restatement, and the Python traceback -- with AWS's +actual sentence buried in a Java stack. Now the driver decides, the same way it does for +failures a worker recorded. +""" + +import pytest + +from python_modules.shared import driver_errors, worker_errors +from python_modules.shared.bulk_executor_error import BulkExecutorError +from python_modules.shared.driver_errors import UNEXPECTED_FAILURE_BANNER, surface + +PY4J_DENIAL = """An error occurred while calling o304.load. +: software.amazon.awssdk.services.dynamodb.model.DynamoDbException: User: \ +arn:aws:sts::1:assumed-role/Role/GlueJobRunnerSession is not authorized to perform: \ +dynamodb:Scan on resource: arn:aws:dynamodb:us-east-1:1:table/t because no \ +identity-based policy allows the dynamodb:Scan action (Service: DynamoDb, Status Code: 400) +\tat software.amazon.awssdk.services.dynamodb.model.DynamoDbException$BuilderImpl.build(DynamoDbException.java:113) +\tat software.amazon.awssdk.core.internal.http.pipeline.stages.RetryableStage.execute(RetryableStage.java:86) +""" + + +def _real_errors_module(): + """Load shared/errors.py from disk. conftest replaces that module with a Mock for the + whole server suite, and str(exception) is not a substitute: the entire point here is + the extraction get_error_message does on a Py4J stack.""" + import importlib.util + from pathlib import Path + + path = Path(__file__).resolve().parents[3] / "server/src/python_modules/shared/errors.py" + spec = importlib.util.spec_from_file_location("_real_errors_for_test", path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +@pytest.fixture(autouse=True) +def real_message_extraction(monkeypatch): + """The classifier binds these at import time, against conftest's Mock.""" + real = _real_errors_module() + monkeypatch.setattr(worker_errors, 'get_error_code', real.get_error_code) + monkeypatch.setattr(worker_errors, 'get_error_message', real.get_error_message) + + +class TestSurfaceUnderstood: + + def test_bulk_executor_error_exits_with_its_sentence(self, capsys): + with pytest.raises(SystemExit) as raised: + surface(BulkExecutorError("Invalid 'where': no such column")) + + assert str(raised.value) == "Invalid 'where': no such column" + assert capsys.readouterr().out == '', "nothing to add; the sentence is the report" + + def test_py4j_denial_exits_with_aws_own_sentence(self, capsys): + """The row that made #332 worth filing: 300 lines of Java, one useful sentence.""" + with pytest.raises(SystemExit) as raised: + surface(Exception(PY4J_DENIAL)) + + reason = str(raised.value) + assert 'is not authorized to perform: dynamodb:Scan' in reason + assert 'at software.amazon' not in reason, "no Java frames in the closing line" + assert 'o304.load' not in reason, "Py4J's own wrapper text is not the problem" + assert capsys.readouterr().out == '', "a denial needs no traceback" + + def test_reason_is_one_line(self): + with pytest.raises(SystemExit) as raised: + surface(BulkExecutorError("first line\nsecond line\n\tindented third")) + + assert str(raised.value) == "first line second line indented third" + + def test_long_reason_is_bounded(self): + with pytest.raises(SystemExit) as raised: + surface(BulkExecutorError("x" * 5000)) + + reason = str(raised.value) + assert len(reason) == driver_errors.MAX_REASON_CHARS + assert reason.endswith('...') + + +class TestSurfaceUnexpected: + + def test_prints_banner_and_traceback_then_exits_one_line(self, capsys): + try: + raise KeyError('pk') + except KeyError as e: + with pytest.raises(SystemExit) as raised: + surface(e) + + out = capsys.readouterr().out + assert UNEXPECTED_FAILURE_BANNER in out + assert 'Traceback' in out and "raise KeyError('pk')" in out, \ + "the frames are the report for something we did not expect" + + reason = str(raised.value) + assert reason == "KeyError: 'pk'", "the type, since a KeyError message is bare" + assert 'Traceback' not in reason, \ + "the reason becomes Glue's ErrorMessage and the client's closing line" + + def test_exits_rather_than_re_raising(self): + """Re-raising is what produced the Glue blob and the Py4J restatement.""" + try: + raise RuntimeError('boom') + except RuntimeError as e: + with pytest.raises(SystemExit): + surface(e) diff --git a/tools/bulk_executor/tests/server/shared/test_errors_message_extraction.py b/tools/bulk_executor/tests/server/shared/test_errors_message_extraction.py new file mode 100644 index 00000000..79152777 --- /dev/null +++ b/tools/bulk_executor/tests/server/shared/test_errors_message_extraction.py @@ -0,0 +1,87 @@ +"""Unit tests for get_error_message's Java/Py4J unwrapping in shared/errors.py. + +A denied connector read reaches the driver as a Py4JJavaError whose str() is the whole +Java stack -- hundreds of frames -- with the sentence the user needs on the second line. +Before #332 nothing extracted it: the SDK v1 pattern in this function stopped matching +when Glue moved to the v2 SDK (`software.amazon.awssdk...DynamoDbException` rather than +`com.amazonaws...AmazonDynamoDBException`), so the "message" was the stack itself. + +shared/errors.py is replaced by a Mock for the whole server suite, so these tests load it +from disk. That is deliberate: the extraction is the behaviour under test. +""" + +import importlib.util +from pathlib import Path + +import pytest + + +@pytest.fixture(scope="module") +def errors(): + path = Path(__file__).resolve().parents[3] / "server/src/python_modules/shared/errors.py" + spec = importlib.util.spec_from_file_location("_real_errors_extraction_test", path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +DENIAL_V2 = """An error occurred while calling o304.load. +: software.amazon.awssdk.services.dynamodb.model.DynamoDbException: User: \ +arn:aws:sts::1:assumed-role/R/GlueJobRunnerSession is not authorized to perform: \ +dynamodb:Scan on resource: arn:aws:dynamodb:us-east-1:1:table/t (Service: DynamoDb, Status Code: 400) +\tat software.amazon.awssdk.services.dynamodb.model.DynamoDbException$BuilderImpl.build(DynamoDbException.java:113) +\tat org.apache.spark.sql.execution.datasources.v2.DataSourceV2Utils$.loadV2Source(DataSourceV2Utils.scala:157) +""" + +WRAPPED_IN_SPARK = """An error occurred while calling o92.save. +: org.apache.spark.SparkException: Job aborted due to stage failure: Task 3 in stage 2.0 failed 4 times +\tat org.apache.spark.scheduler.DAGScheduler.failJobAndIndependentStages(DAGScheduler.scala:2905) +Caused by: software.amazon.awssdk.services.dynamodb.model.DynamoDbException: User: R is not \ +authorized to perform: dynamodb:BatchWriteItem on resource: table/t +\tat software.amazon.awssdk.core.internal.http.pipeline.stages.RetryableStage.execute(RetryableStage.java:86) +""" + +MULTILINE_DETAIL = """An error occurred while calling o1.load. +: java.lang.IllegalArgumentException: Unsupported option 'dynamodb.throughput.read.percent' + supplied for this connector version; use dynamodb.throughput.read instead +\tat com.amazonaws.services.glue.connectors.DynamoDbOptions.validate(DynamoDbOptions.scala:88) +""" + + +class TestJavaErrorUnwrapping: + + def test_sdk_v2_denial_yields_aws_sentence(self, errors): + message = errors.get_error_message(Exception(DENIAL_V2)) + assert message.startswith('User: arn:aws:sts::1:assumed-role/R/GlueJobRunnerSession') + assert 'dynamodb:Scan' in message + assert 'at software.amazon' not in message and 'at org.apache.spark' not in message + assert 'o304.load' not in message, "Py4J's wrapper line is not the error" + + def test_innermost_cause_wins_over_spark_boilerplate(self, errors): + """The outer layer says "Job aborted due to stage failure"; the cause says why.""" + message = errors.get_error_message(Exception(WRAPPED_IN_SPARK)) + assert 'not authorized to perform: dynamodb:BatchWriteItem' in message + assert 'Job aborted due to stage failure' not in message + + def test_message_continuation_lines_are_joined(self, errors): + message = errors.get_error_message(Exception(MULTILINE_DETAIL)) + assert message == ( + "Unsupported option 'dynamodb.throughput.read.percent' supplied for this " + "connector version; use dynamodb.throughput.read instead" + ) + + def test_sdk_v1_pattern_still_works(self, errors): + """Glue 4 and earlier; kept so an older runtime does not regress.""" + v1 = ("com.amazonaws.services.dynamodbv2.model.AmazonDynamoDBException: " + "Requested resource not found (Service: AmazonDynamoDBv2; Status Code: 400)") + assert errors.get_error_message(Exception(v1)) == "Requested resource not found" + + def test_a_plain_exception_is_unchanged(self, errors): + assert errors.get_error_message(ValueError("just a message")) == "just a message" + + def test_boto_error_response_still_preferred(self, errors): + """An AWS SDK error carries its own message; do not go looking in the string.""" + class Fake(Exception): + response = {'Error': {'Code': 'AccessDeniedException', 'Message': 'denied by policy'}} + + assert errors.get_error_message(Fake()) == 'denied by policy' diff --git a/tools/bulk_executor/tests/server/test_load.py b/tools/bulk_executor/tests/server/test_load.py index 96107c54..c78274ab 100644 --- a/tools/bulk_executor/tests/server/test_load.py +++ b/tools/bulk_executor/tests/server/test_load.py @@ -274,7 +274,8 @@ def test_returns_early_when_count_is_zero(self, monkeypatch): assert result is None def test_raises_on_count_exception(self, monkeypatch): - """Line 99-100: exception during count() re-raises wrapped.""" + """A failure reading the source is the user's to fix -- wrong --format, malformed + data -- so it reports as a sentence naming the path and format (#332).""" monkeypatch.setattr(load_module, 'check_s3_file_exists', lambda uri: True) monkeypatch.setattr(load_module, 'get_dynamodb_throughput_configs', lambda *a, **kw: {}) @@ -282,9 +283,12 @@ def test_raises_on_count_exception(self, monkeypatch): df.count.side_effect = RuntimeError('spark error') monkeypatch.setattr(load_module, 'read_data', lambda *a: df) - with pytest.raises(Exception, match="Failed to create DynamicFrame"): + with pytest.raises(load_module.BulkExecutorError, match="Could not read the source") as exc: load_module.run(MagicMock(), MagicMock(), MagicMock(), {'table': 't', 's3_path': 's3://b/k', 'format': 'csv'}) + assert "s3://b/k" in str(exc.value) and "'csv'" in str(exc.value), ( + "name the path and the format the user claimed it was" + ) class TestRunRemoveEmptyStrings: diff --git a/tools/bulk_executor/tests/server/test_root.py b/tools/bulk_executor/tests/server/test_root.py index cc04f7c9..e309e399 100644 --- a/tools/bulk_executor/tests/server/test_root.py +++ b/tools/bulk_executor/tests/server/test_root.py @@ -113,12 +113,18 @@ def _install_logger_stub(modules_to_install): def _install_bulk_executor_error_stub(modules_to_install): - """Provide a real exception class so `except BulkExecutorError` actually matches.""" - be_module = types.ModuleType("python_modules.shared.bulk_executor_error") - - class BulkExecutorError(Exception): - pass + """Expose the class the classifier itself holds, under the stubbed module path. + + Not a fresh subclass. root.py now reports through shared/driver_errors.py, which asks + shared/worker_errors.py whether the failure is understood -- and that check is an + isinstance against the class *it* bound at import time. conftest registers both + `shared.` and `python_modules.shared.` over the same files, so importing the module + again can produce a second, distinct class; the isinstance would then miss and the + test would assert nothing about the branch it names. + """ + from python_modules.shared.worker_errors import BulkExecutorError + be_module = types.ModuleType("python_modules.shared.bulk_executor_error") be_module.BulkExecutorError = BulkExecutorError modules_to_install["python_modules.shared.bulk_executor_error"] = be_module return BulkExecutorError @@ -144,6 +150,23 @@ def _load_root(monkeypatch, argv, verb_module=None, verb_name=None, BulkExecutorError = _install_bulk_executor_error_stub(install) + # tests/server/conftest.py replaces shared.errors with a Mock, so get_error_message + # would hand back a Mock repr and any assertion on the message text would be vacuous. + # driver_errors needs a working one to build the closing line. + errors_module = types.ModuleType("python_modules.shared.errors") + errors_module.get_error_message = lambda e: str(e) + errors_module.get_error_code = lambda e: None + errors_module.ListAccumulator = object + install["python_modules.shared.errors"] = errors_module + + # shared/worker_errors.py holds the classifier root.py's reporting path uses, and it + # binds these two names at import time -- which already happened, against conftest's + # Mock. Installing a module above does not rebind them, so patch them where they are + # read or every assertion on message text is vacuous. + import python_modules.shared.worker_errors as _worker_errors + monkeypatch.setattr(_worker_errors, "get_error_message", lambda e: str(e)) + monkeypatch.setattr(_worker_errors, "get_error_code", lambda e: None) + # The verb module is what root.py imports via importlib.import_module. # If `import_should_fail` is set, we leave python_modules. # uninstalled and patch importlib.import_module to raise ImportError so @@ -581,23 +604,38 @@ def test_bulk_executor_error_exits_cleanly(self, monkeypatch): class TestRootGenericExceptionPropagation: - """Generic Exception from a verb is re-raised (lines 88-89).""" + """A non-BulkExecutorError from a verb is reported, then exits (#332). - def test_generic_exception_propagates(self, monkeypatch): - """Lines 88-89: a non-BulkExecutorError exception is re-raised verbatim.""" + It used to be re-raised, which handed the user Glue's exception-analysis blob and + Py4J's restatement of the same failure on top of the traceback. root.py now routes + everything through shared/driver_errors.py: the traceback is printed once for the + user, and the job exits with a one-line reason that becomes Glue's ErrorMessage. + """ + + def test_generic_exception_is_reported_then_exits(self, monkeypatch, capsys): run_mock = MagicMock(side_effect=RuntimeError("boom")) verb = _make_verb_module("copy", run_callable=run_mock) - with pytest.raises(RuntimeError, match="boom"): + with pytest.raises(SystemExit) as exc_info: _load_root(monkeypatch, ["root.py", "--XAction", "copy"], verb_module=verb, verb_name="copy") + reason = str(exc_info.value) + assert "boom" in reason, "the closing line names the failure" + assert "Traceback" not in reason, ( + "the reason is Glue's ErrorMessage and the client's last line -- keep it to one" + ) + out = capsys.readouterr().out + assert "did not expect" in out and "Traceback" in out, ( + "an unexpected failure prints its traceback where the user will see it" + ) + def test_generic_exception_skips_commit_and_stop(self, monkeypatch): - """Lines 93-94: exception path bypasses job.commit() and spark.stop().""" + """A failed job must not be committed -- that is what makes Glue mark it FAILED.""" run_mock = MagicMock(side_effect=ValueError("nope")) verb = _make_verb_module("copy", run_callable=run_mock) awsglue = _build_awsglue_stubs() - with pytest.raises(ValueError): + with pytest.raises(SystemExit): _load_root(monkeypatch, ["root.py", "--XAction", "copy"], verb_module=verb, verb_name="copy", diff --git a/tools/bulk_executor/tests/server/test_sql.py b/tools/bulk_executor/tests/server/test_sql.py index 190cad97..cedd0382 100644 --- a/tools/bulk_executor/tests/server/test_sql.py +++ b/tools/bulk_executor/tests/server/test_sql.py @@ -424,9 +424,11 @@ def test_valid_limit_runtime_failure_wraps_as_sql_query_error(self, monkeypatch, mock_table_info, mock_spark_session, mock_get_error_message, glue_context, base_args): - """A valid integer limit that then fails in Spark is a runtime error, not a - bad-parameter error: it flows to the outer 'SQL query error' handler rather - than being mislabeled 'Invalid limit'.""" + """A valid integer limit that then fails in Spark still reports as a query error + rather than being mislabeled 'Invalid limit'. + + Since #332 both are BulkExecutorError -- a query is the user's to fix either way, + so neither deserves a stack trace. The distinction that matters is the message.""" base_args['limit'] = '5' result = MagicMock() result.limit.side_effect = RuntimeError("spark limit failure") @@ -434,10 +436,11 @@ def test_valid_limit_runtime_failure_wraps_as_sql_query_error(self, monkeypatch, df = MagicMock() glue_context.create_dynamic_frame.from_options.return_value.toDF.return_value = df - with pytest.raises(Exception, match="SQL query error") as exc: + with pytest.raises(sql_module.BulkExecutorError, match="SQL query error") as exc: sql_module.run(MagicMock(), MagicMock(), glue_context, base_args) - # It is NOT a clean user-parameter error. - assert not isinstance(exc.value, sql_module.BulkExecutorError) + assert "Invalid 'limit'" not in str(exc.value), ( + "a Spark-side failure must not be reported as a bad --limit" + ) mock_get_error_message.assert_called() diff --git a/tools/bulk_executor/tests/server/test_worker_failures_go_through_worker_errors.py b/tools/bulk_executor/tests/server/test_worker_failures_go_through_worker_errors.py index ac5d2433..76e743c6 100644 --- a/tools/bulk_executor/tests/server/test_worker_failures_go_through_worker_errors.py +++ b/tools/bulk_executor/tests/server/test_worker_failures_go_through_worker_errors.py @@ -18,6 +18,8 @@ """ import re + +import pytest from pathlib import Path _REPO = Path(__file__).resolve().parents[2] @@ -89,16 +91,18 @@ def test_the_seam_is_actually_used(): f"expected several verbs to surface worker failures, found {surfacing}") -def test_the_client_watches_for_the_banner_the_job_prints(): +@pytest.mark.parametrize("module", ["worker_errors.py", "driver_errors.py"]) +def test_the_client_watches_for_the_banners_the_job_prints(module): """The client suppresses Glue's exception-analysis noise once the job has explained - itself. That handshake is two string literals in two trees: if they drift, an - unexpected failure gets the worker traceback *and* a Glue traceback of our plumbing - on top of it, which is exactly what the banner exists to prevent.""" - server = (_SERVER_SRC / "python_modules" / "shared" / "worker_errors.py").read_text() + itself. That handshake is string literals in two trees: if they drift, an unexpected + failure gets our traceback *and* Glue's blob on top of it, which is exactly what the + banner exists to prevent. Both sides print one -- a worker failure and a driver + failure -- so both are checked.""" + server = (_SERVER_SRC / "python_modules" / "shared" / module).read_text() banner = re.search(r'UNEXPECTED_FAILURE_BANNER = "([^"]+)"', server) - assert banner, "worker_errors.py no longer defines UNEXPECTED_FAILURE_BANNER" + assert banner, f"{module} no longer defines UNEXPECTED_FAILURE_BANNER" client = (_REPO / "client" / "src" / "runner.py").read_text() assert banner.group(1) in client, ( f"client/src/runner.py does not watch for {banner.group(1)!r}; " - "JOB_EXPLAINED_THE_FAILURE has drifted from worker_errors.py") + f"JOB_EXPLAINED_THE_FAILURE has drifted from {module}") From 16543102f911983385cc7e72eaa1acfe10ba882a Mon Sep 17 00:00:00 2001 From: Jason Hunter Date: Tue, 1 Sep 2026 01:58:58 -0700 Subject: [PATCH 2/8] [bulk] Mark an understood driver failure so Glue's blob is suppressed every time Found in the live before/after: a denied `find` printed our clean sentence and then 183 lines of Glue exception analysis anyway, while a denied `count` in the same batch printed none. Glue emits that blob after a clean sys.exit only sometimes, and the client can only drop it once the job has identified itself -- which previously happened only when the exception was literally a BulkExecutorError. Understood driver failures now log behind EXPLAINED_FAILURE_PREFIX ("Failure: "), the client watches for it, and the drift guard checks all three markers rather than one. Asserted client-side as well, since whether the blob shows up in any given live run is not something a test can rely on. --- tools/bulk_executor/client/src/runner.py | 1 + .../src/python_modules/fill/alwaysboom.py | 5 ++++ .../python_modules/shared/driver_errors.py | 11 +++++++- tools/bulk_executor/server/src/root.py | 7 +++--- .../bulk_executor/tests/client/test_runner.py | 25 +++++++++++++++++++ ...orker_failures_go_through_worker_errors.py | 19 ++++++++------ 6 files changed, 57 insertions(+), 11 deletions(-) create mode 100644 tools/bulk_executor/server/src/python_modules/fill/alwaysboom.py diff --git a/tools/bulk_executor/client/src/runner.py b/tools/bulk_executor/client/src/runner.py index 1c12ca04..ba0106f9 100644 --- a/tools/bulk_executor/client/src/runner.py +++ b/tools/bulk_executor/client/src/runner.py @@ -64,6 +64,7 @@ 'BulkExecutorError', 'A worker failed in a way we did not expect. Traceback from the worker:', 'The job failed in a way we did not expect. Traceback:', + 'Failure: ', ) # Timing constants diff --git a/tools/bulk_executor/server/src/python_modules/fill/alwaysboom.py b/tools/bulk_executor/server/src/python_modules/fill/alwaysboom.py new file mode 100644 index 00000000..d0045cf7 --- /dev/null +++ b/tools/bulk_executor/server/src/python_modules/fill/alwaysboom.py @@ -0,0 +1,5 @@ +"""Scratch generator for #332: raises on every call, so it fails during the driver's size peek.""" + + +def generate(): + raise RuntimeError("faker exploded on every call") diff --git a/tools/bulk_executor/server/src/python_modules/shared/driver_errors.py b/tools/bulk_executor/server/src/python_modules/shared/driver_errors.py index a47f6a55..13866a0a 100644 --- a/tools/bulk_executor/server/src/python_modules/shared/driver_errors.py +++ b/tools/bulk_executor/server/src/python_modules/shared/driver_errors.py @@ -27,6 +27,13 @@ UNEXPECTED_FAILURE_BANNER = "The job failed in a way we did not expect. Traceback:" +# Printed in front of an understood failure's sentence. Two jobs: it tells the reader this +# is the explanation rather than one more log line, and the client watches for it to +# suppress Glue's exception-analysis blob. Glue emits that blob for a clean sys.exit only +# sometimes -- observed on a denied `find` but not a denied `count` in the same batch -- so +# the marker has to be there every time, not only when a BulkExecutorError was involved. +EXPLAINED_FAILURE_PREFIX = "Failure: " + # Cap on what we hand to sys.exit(): Glue records it as the job's ErrorMessage and the # client prints it as its closing line. AWS's authorization sentences run ~400 chars, # which is worth keeping whole; a Java stack that slipped through is not. @@ -62,7 +69,9 @@ def surface(exception): reason = _one_line(reason) if isinstance(exception, BulkExecutorError): + # Keep the name users have seen since before any of this, and the client's + # long-standing suppression marker. log.error(f"BulkExecutorError: {reason}") else: - log.error(reason) + log.error(f"{EXPLAINED_FAILURE_PREFIX}{reason}") sys.exit(reason) diff --git a/tools/bulk_executor/server/src/root.py b/tools/bulk_executor/server/src/root.py index beb59902..af2ce9e6 100644 --- a/tools/bulk_executor/server/src/root.py +++ b/tools/bulk_executor/server/src/root.py @@ -112,9 +112,10 @@ def _get_parsed_glue_job_args(argv): # hands the user Glue's exception-analysis blob plus Py4J's restatement of the # same error, with AWS's actual sentence buried inside a Java stack (#332). # - # BaseException, not Exception: a worker calling exit() reaches the driver as a - # SystemExit, and Py4J wraps KeyboardInterrupt-style aborts too. sys.exit from - # driver_errors.surface() raises SystemExit itself, so let that through. + # BaseException rather than Exception so nothing escapes by inheriting from + # the wrong base -- but SystemExit passes straight through: a helper that + # already called exit() has said its piece, and re-reporting it would relabel + # a deliberate exit as a surprise. if isinstance(e, SystemExit): raise driver_errors.surface(e) diff --git a/tools/bulk_executor/tests/client/test_runner.py b/tools/bulk_executor/tests/client/test_runner.py index 7b93559b..ec4414df 100644 --- a/tools/bulk_executor/tests/client/test_runner.py +++ b/tools/bulk_executor/tests/client/test_runner.py @@ -345,6 +345,31 @@ def test_suppresses_glue_metrics_reporter_stack(self, bulk_runner, capsys): assert captured.out == '' assert captured.err == '' + def test_driver_failure_marker_suppresses_the_glue_blob(self, bulk_runner, capsys, monkeypatch): + """#332: an understood driver-side failure prints one sentence, and Glue's + exception-analysis blob that sometimes follows a clean sys.exit is dropped. + + Glue emits that blob unpredictably -- observed after a denied `find` but not after a + denied `count` in the same batch -- so this is asserted here rather than trusted to + show up in a live run.""" + monkeypatch.setattr(runner_module.utils, 'CONFIG_LOG_MESSAGE_KEYS', []) + monkeypatch.setattr(runner_module.utils, 'STD_ERROR_MESSAGE_KEYS', []) + + denial = _make_event(message=( + "Failure: User: arn:aws:sts::1:assumed-role/R/GlueJobRunnerSession is not " + "authorized to perform: dynamodb:Scan on resource: table/t" + )) + bulk_runner._pretty_print_log_event(denial) + first = capsys.readouterr().out + assert 'not authorized to perform' in first, "the sentence itself must print" + + blob = _make_event(message=( + "2026-09-01 08:49:19 ERROR GlueExceptionAnalysisListener:9 - " + "[Glue Exception Analysis] {\"Failure Reason\": \"Traceback (most recent call last)...\"}" + )) + bulk_runner._pretty_print_log_event(blob) + assert capsys.readouterr().out == '', "Glue's restatement adds nothing after it" + def test_real_worker_traceback_still_prints(self, bulk_runner, capsys): """Guard for #334: the anchor is Glue's reporter, not the frames. An unexpected worker failure prints its traceback through the same path and must survive.""" diff --git a/tools/bulk_executor/tests/server/test_worker_failures_go_through_worker_errors.py b/tools/bulk_executor/tests/server/test_worker_failures_go_through_worker_errors.py index 76e743c6..44f8563e 100644 --- a/tools/bulk_executor/tests/server/test_worker_failures_go_through_worker_errors.py +++ b/tools/bulk_executor/tests/server/test_worker_failures_go_through_worker_errors.py @@ -91,18 +91,23 @@ def test_the_seam_is_actually_used(): f"expected several verbs to surface worker failures, found {surfacing}") -@pytest.mark.parametrize("module", ["worker_errors.py", "driver_errors.py"]) -def test_the_client_watches_for_the_banners_the_job_prints(module): +@pytest.mark.parametrize("constant", [ + ("worker_errors.py", "UNEXPECTED_FAILURE_BANNER"), + ("driver_errors.py", "UNEXPECTED_FAILURE_BANNER"), + ("driver_errors.py", "EXPLAINED_FAILURE_PREFIX"), +]) +def test_the_client_watches_for_every_marker_the_job_prints(constant): """The client suppresses Glue's exception-analysis noise once the job has explained itself. That handshake is string literals in two trees: if they drift, an unexpected failure gets our traceback *and* Glue's blob on top of it, which is exactly what the banner exists to prevent. Both sides print one -- a worker failure and a driver failure -- so both are checked.""" + module, name = constant server = (_SERVER_SRC / "python_modules" / "shared" / module).read_text() - banner = re.search(r'UNEXPECTED_FAILURE_BANNER = "([^"]+)"', server) - assert banner, f"{module} no longer defines UNEXPECTED_FAILURE_BANNER" + marker = re.search(rf'{name} = "([^"]+)"', server) + assert marker, f"{module} no longer defines {name}" client = (_REPO / "client" / "src" / "runner.py").read_text() - assert banner.group(1) in client, ( - f"client/src/runner.py does not watch for {banner.group(1)!r}; " - f"JOB_EXPLAINED_THE_FAILURE has drifted from {module}") + assert marker.group(1) in client, ( + f"client/src/runner.py does not watch for {marker.group(1)!r}; " + f"JOB_EXPLAINED_THE_FAILURE has drifted from {module}.{name}") From 3b63be5bf10eb994865ffe0d5c41706c55a02c0b Mon Sep 17 00:00:00 2001 From: Jason Hunter Date: Tue, 1 Sep 2026 01:59:41 -0700 Subject: [PATCH 3/8] [bulk] Assert the server emits the understood-failure marker, not just that the client honours it --- .../tests/server/shared/test_driver_errors.py | 25 +++++++++++++++++++ 1 file changed, 25 insertions(+) diff --git a/tools/bulk_executor/tests/server/shared/test_driver_errors.py b/tools/bulk_executor/tests/server/shared/test_driver_errors.py index 09252dbe..483eaa01 100644 --- a/tools/bulk_executor/tests/server/shared/test_driver_errors.py +++ b/tools/bulk_executor/tests/server/shared/test_driver_errors.py @@ -69,6 +69,31 @@ def test_py4j_denial_exits_with_aws_own_sentence(self, capsys): assert 'o304.load' not in reason, "Py4J's own wrapper text is not the problem" assert capsys.readouterr().out == '', "a denial needs no traceback" + def test_logs_behind_the_marker_the_client_watches(self, caplog): + """The sentence must carry EXPLAINED_FAILURE_PREFIX. Without it the client cannot + tell the job has explained itself, and Glue's exception-analysis blob -- which + follows a clean exit only sometimes -- reaches the user anyway.""" + import logging + + with caplog.at_level(logging.ERROR): + with pytest.raises(SystemExit): + surface(Exception(PY4J_DENIAL)) + + logged = ' '.join(r.message for r in caplog.records) + assert driver_errors.EXPLAINED_FAILURE_PREFIX in logged + assert 'is not authorized' in logged + + def test_bulk_executor_error_keeps_its_own_marker(self, caplog): + """The name users have seen for years, and the client's original marker.""" + import logging + + with caplog.at_level(logging.ERROR): + with pytest.raises(SystemExit): + surface(BulkExecutorError("PITR must be enabled")) + + logged = ' '.join(r.message for r in caplog.records) + assert 'BulkExecutorError: PITR must be enabled' in logged + def test_reason_is_one_line(self): with pytest.raises(SystemExit) as raised: surface(BulkExecutorError("first line\nsecond line\n\tindented third")) From b4e40f0c831b476dc46888d4a6eca6fb960cdc71 Mon Sep 17 00:00:00 2001 From: Jason Hunter Date: Tue, 1 Sep 2026 02:29:05 -0700 Subject: [PATCH 4/8] [bulk] Report a format mismatch as the user's mistake wherever Spark notices it The live before/after caught this: `--format parquet` at a CSV file was reported as an unexpected failure ("Py4JJavaError: ... is not a Parquet file", with a traceback) rather than as the user's own mistake. Parquet reads its footer while the DynamicFrame is being created, so it raises from read_data -- outside the handler that already converted the count() path, which is where a JSON or CSV mismatch surfaces. read_data now sits inside that handler, so both report the same way: one sentence naming the path and the format the user claimed it was. Also updates the lint rule's invariant 2 for #332: the net exists, a marker has to reach the console (Glue's blob follows a clean exit only sometimes -- 183 lines after a denied find, none after a denied count in the same batch), and get_error_message has to unwrap the Java stack to the innermost cause. --- .../rules/dynamodb_failures_reported.md | 47 ++++++++++++------- .../src/python_modules/load/__init__.py | 8 +++- tools/bulk_executor/tests/server/test_load.py | 17 +++++++ 3 files changed, 54 insertions(+), 18 deletions(-) diff --git a/tools/bulk_executor/ai_lint/rules/dynamodb_failures_reported.md b/tools/bulk_executor/ai_lint/rules/dynamodb_failures_reported.md index c236deea..e3507a77 100644 --- a/tools/bulk_executor/ai_lint/rules/dynamodb_failures_reported.md +++ b/tools/bulk_executor/ai_lint/rules/dynamodb_failures_reported.md @@ -148,15 +148,32 @@ handler exists or is broad enough. ## Invariant 2 — driver-side work fails politely too A verb whose DynamoDB access happens on the **driver** (`find`, `count`, `sql` read -through `shared/glue_connector`'s `.load()`) has no worker code to check, but it is -not therefore exempt. A denied `.load()` must still reach the user as one sentence. - -The channel is `BulkExecutorError`: `root.py` catches it and calls -`sys.exit(str(e))`, Glue records that as the job's `ErrorMessage`, and the client -prints it as its closing line. The PITR guard already rides this path and produces -exactly one clean sentence. Anything else — a bare Spark/Py4J exception, or a plain -`Exception` — leaves the user a traceback prefixed with Glue's error category, even -when the text underneath is perfectly good. +through `shared/glue_connector`'s `.load()`; `load` writes through it) has no worker code +to check, but it is not therefore exempt. A denied `.load()` must still reach the user as +one sentence. + +Since #332 there is a net: `root.py` routes everything out of a verb through +`shared/driver_errors.py`, which classifies it the same way workers are classified and +**exits rather than re-raising**. Re-raising is what summons Glue's exception-analysis +blob and Py4J's restatement of the same failure. So a verb no longer *has* to convert its +own exception for the user to get a sentence — but converting is still better where the +verb knows something the net cannot, which is why `sql`, `load` and the transform loader +raise `BulkExecutorError` with their own wording. + +Two things to check when reading driver-side code: + +- **A marker must reach the console.** An understood failure logs behind + `driver_errors.EXPLAINED_FAILURE_PREFIX`, an unexpected one behind + `driver_errors.UNEXPECTED_FAILURE_BANNER`, and the client suppresses Glue's blob when it + sees either (or `BulkExecutorError`). Glue emits that blob after a clean `sys.exit` only + *sometimes* — measured: a denied `find` produced 183 blob lines while a denied `count` + in the same batch produced none — so a missing marker is a bug you will only see half + the time. +- **The message has to survive the Java stack.** A Py4J error's `str()` is the whole + stack. `get_error_message` unwraps it to the innermost cause, because the outer layer is + usually Spark boilerplate ("Job aborted due to stage failure") while the cause is AWS's + sentence. If you see a closing line reading `An error occurred while calling o304.load`, + that unwrapping did not happen. Two things that look like compliance and are not: @@ -280,13 +297,11 @@ State what you verified even when clean, so a pass is trustworthy. Outputs are kept in `~/Documents/bulk-331-runs/` with `before/` counterparts on `main` (597-668 lines each). -- **Does not conform to invariant 2, tracked as #332:** `find`, `count` and `sql`. - A table the role cannot `Scan` gives 314-324 lines closing on - `Error Category: UNCLASSIFIED_ERROR; Failed Line Number: 1362; An error occurred - while calling o304.load. User: ... is not authorized ...`. AWS's sentence is in - there, but it arrives as an unhandled Py4J exception. Compare the worker-side - verbs, which now close on `Error during delete: User ... is not authorized to - perform: dynamodb:BatchWriteItem` in 26-82 lines. +- Invariant 2 conforms as of #332, via the `root.py` net plus per-site wording in `sql`, + `load` and `transform_loader`. Before it, a table the role could not `Scan` gave + 314-324 lines closing on `Error Category: UNCLASSIFIED_ERROR; ...; An error occurred + while calling o304.load. User: ... is not authorized ...`; the sentence was in there, + but it arrived as an unhandled Py4J exception. - Partly audited: `load`, `load_export`, `revert_export`. Their shared write path (`shared/export/pipeline/writer.py`) is now correct, but whether every worker entry point in those paths records rather than escapes has not been checked — treat the diff --git a/tools/bulk_executor/server/src/python_modules/load/__init__.py b/tools/bulk_executor/server/src/python_modules/load/__init__.py index 845bb87b..9e9d18d4 100644 --- a/tools/bulk_executor/server/src/python_modules/load/__init__.py +++ b/tools/bulk_executor/server/src/python_modules/load/__init__.py @@ -81,10 +81,14 @@ def run(job, spark_context, glue_context, parsed_args): if not check_s3_file_exists(s3_path): raise BulkExecutorError(f"The S3 path '{s3_path}' doesn't exist or is not accessible") - dynamicFrame = read_data(glue_context, s3_path, parsed_args) - + # Inside the same handler as the count below: some formats fail here instead. Parquet + # reads its footer while the frame is being created, so `--format parquet` at a CSV + # file raises from read_data, while a JSON or CSV mismatch only surfaces at count(). + # Measured before this was wrapped: the Parquet case reported as an unexpected failure + # ("Py4JJavaError: ... is not a Parquet file") rather than as the user's own mistake. count = 0 try: + dynamicFrame = read_data(glue_context, s3_path, parsed_args) count = dynamicFrame.count() if count == 0: log.error("No data found, please check your data source") # Should perhaps check that the path exists diff --git a/tools/bulk_executor/tests/server/test_load.py b/tools/bulk_executor/tests/server/test_load.py index c78274ab..5ba9318a 100644 --- a/tools/bulk_executor/tests/server/test_load.py +++ b/tools/bulk_executor/tests/server/test_load.py @@ -273,6 +273,23 @@ def test_returns_early_when_count_is_zero(self, monkeypatch): {'table': 't', 's3_path': 's3://b/k', 'format': 'csv'}) assert result is None + def test_raises_when_the_frame_cannot_even_be_created(self, monkeypatch): + """Parquet reads its footer while the frame is created, so `--format parquet` at a + CSV file raises from read_data rather than from count(). Both are the same mistake + and must read the same way; measured live before this was wrapped, the Parquet case + came out as "Py4JJavaError: ... is not a Parquet file" with a traceback.""" + monkeypatch.setattr(load_module, 'check_s3_file_exists', lambda uri: True) + + def boom(*_args): + raise RuntimeError("s3://b/k/data.csv is not a Parquet file") + + monkeypatch.setattr(load_module, 'read_data', boom) + + with pytest.raises(load_module.BulkExecutorError, match="Could not read the source") as exc: + load_module.run(MagicMock(), MagicMock(), MagicMock(), + {'table': 't', 's3_path': 's3://b/k/data.csv', 'format': 'parquet'}) + assert "'parquet'" in str(exc.value), "name the format the user claimed" + def test_raises_on_count_exception(self, monkeypatch): """A failure reading the source is the user's to fix -- wrong --format, malformed data -- so it reports as a sentence naming the path and format (#332).""" From f804d1a09ac3062cfa9c236beec5abebe1265264 Mon Sep 17 00:00:00 2001 From: Jason Hunter Date: Tue, 1 Sep 2026 02:33:39 -0700 Subject: [PATCH 5/8] [bulk] Drop Spark's own duplicate dump of a query-analysis failure Measured while verifying #332: `sql` with a mistyped column produced 148 lines, 90 of them Spark logging the AnalysisException itself -- at ERROR, before our handler turns it into "SQL query error: [UNRESOLVED_COLUMN...]" -- as one 10,850-char JSON event carrying the message we already print, ~90 Java frames, and the unresolved query plan. One event, so one ignore-list anchor drops all of it, checked against the raw CloudWatch event rather than assumed. Anchored on the logger name inside the JSON; a guard test covers the tempting alternative of anchoring on the error text, which would have hidden the verb's own sentence too. --- .../client/src/utils/__init__.py | 7 ++++++ .../bulk_executor/tests/client/test_runner.py | 24 +++++++++++++++++++ 2 files changed, 31 insertions(+) diff --git a/tools/bulk_executor/client/src/utils/__init__.py b/tools/bulk_executor/client/src/utils/__init__.py index 80c34d92..a4b92a8a 100644 --- a/tools/bulk_executor/client/src/utils/__init__.py +++ b/tools/bulk_executor/client/src/utils/__init__.py @@ -38,6 +38,13 @@ # aws-glue-di-package.jar or metrics-core. Remove once AWS fixes the image. # Issue #334. r"Exception thrown from AWSDILyraMetricsReporter#report", + # Spark reports a query-analysis failure itself, at ERROR, before our handler sees it: + # one ~10 KB JSON event carrying the message we go on to print cleanly, plus ~90 Java + # frames and the unresolved query plan. Measured on `sql` with a mistyped column: 90 of + # the run's 148 lines. Anchored on the logger name inside that JSON, so it drops the + # whole event rather than leaving orphaned frames. The message still reaches the user + # through the verb's own "SQL query error: ..." line. Issue #332. + r'"logger": "SQLQueryContextLogger"', ] # Intentional nuanced configs: diff --git a/tools/bulk_executor/tests/client/test_runner.py b/tools/bulk_executor/tests/client/test_runner.py index ec4414df..f1093ad1 100644 --- a/tools/bulk_executor/tests/client/test_runner.py +++ b/tools/bulk_executor/tests/client/test_runner.py @@ -370,6 +370,30 @@ def test_driver_failure_marker_suppresses_the_glue_blob(self, bulk_runner, capsy bulk_runner._pretty_print_log_event(blob) assert capsys.readouterr().out == '', "Glue's restatement adds nothing after it" + def test_suppresses_sparks_own_query_analysis_dump(self, bulk_runner, capsys): + """#332: Spark logs an analysis failure itself, at ERROR, as one JSON event holding + the message plus ~90 Java frames and the query plan -- before our handler turns it + into "SQL query error: ...". Measured on a mistyped column: 90 of 148 lines. + + The message below is the head of the verbatim CloudWatch event (10,850 chars, a + single event, so one anchor drops all of it). + """ + ev = _make_event(message="{\"ts\": \"2026-09-01 09:31:40.589\", \"level\": \"ERROR\", \"logger\": \"SQLQueryContextLogger\", \"msg\": \"[UNRESOLVED_COLUMN.WITH_SUGGESTION] A column, variable, or function parameter with name `nosuchcolumn` cannot be resolved. Did you mean one of the following? [`payload`, `sk`, `pk`]. SQLSTATE: 42703\", \"context\": {\"errorClass\": \"UNRESOLVED_COLUMN.WITH_SUGGESTION\"}, \"exception\": {\"class\": \"Py4JJavaError\", ") + bulk_runner._pretty_print_log_event(ev) + captured = capsys.readouterr() + assert captured.out == '' and captured.err == '' + + def test_our_own_sql_query_error_still_prints(self, bulk_runner, capsys): + """Guard: the anchor is Spark's logger name, not the error text, so the sentence the + verb produces is unaffected.""" + ev = _make_event(message=( + "2026-09-01 09:31:41,173 ERROR - BulkExecutorError: SQL query error: " + "[UNRESOLVED_COLUMN.WITH_SUGGESTION] A column with name `nosuchcolumn` cannot be resolved" + )) + bulk_runner._pretty_print_log_event(ev) + captured = capsys.readouterr() + assert 'UNRESOLVED_COLUMN' in captured.out + captured.err + def test_real_worker_traceback_still_prints(self, bulk_runner, capsys): """Guard for #334: the anchor is Glue's reporter, not the frames. An unexpected worker failure prints its traceback through the same path and must survive.""" From 91e2c94d1b11341794cc86f4337bdfb1b4ccf9e7 Mon Sep 17 00:00:00 2001 From: Jason Hunter Date: Tue, 1 Sep 2026 03:07:27 -0700 Subject: [PATCH 6/8] [bulk] Cover the driver reporting paths, and drop the scratch generator I committed fill/alwaysboom.py was a throwaway generator for reproducing a driver-side failure and should never have been committed -- removed. Coverage additions, all reachable now that there is a way to import shared/errors.py in a suite that stubs it: Spark's ParseException branch (message plus the offending SQL), the .message-attribute branch, and root.py's SystemExit passthrough. The first two were pre-existing gaps; the third is mine. shared/errors.py goes 54% -> 73%, root.py 96% -> 99%. --- .../src/python_modules/fill/alwaysboom.py | 5 -- .../shared/test_errors_message_extraction.py | 55 ++++++++++++++++--- tools/bulk_executor/tests/server/test_root.py | 13 +++++ 3 files changed, 60 insertions(+), 13 deletions(-) delete mode 100644 tools/bulk_executor/server/src/python_modules/fill/alwaysboom.py diff --git a/tools/bulk_executor/server/src/python_modules/fill/alwaysboom.py b/tools/bulk_executor/server/src/python_modules/fill/alwaysboom.py deleted file mode 100644 index d0045cf7..00000000 --- a/tools/bulk_executor/server/src/python_modules/fill/alwaysboom.py +++ /dev/null @@ -1,5 +0,0 @@ -"""Scratch generator for #332: raises on every call, so it fails during the driver's size peek.""" - - -def generate(): - raise RuntimeError("faker exploded on every call") diff --git a/tools/bulk_executor/tests/server/shared/test_errors_message_extraction.py b/tools/bulk_executor/tests/server/shared/test_errors_message_extraction.py index 79152777..92e5e25c 100644 --- a/tools/bulk_executor/tests/server/shared/test_errors_message_extraction.py +++ b/tools/bulk_executor/tests/server/shared/test_errors_message_extraction.py @@ -10,19 +10,31 @@ from disk. That is deliberate: the extraction is the behaviour under test. """ -import importlib.util -from pathlib import Path +import importlib import pytest -@pytest.fixture(scope="module") +@pytest.fixture def errors(): - path = Path(__file__).resolve().parents[3] / "server/src/python_modules/shared/errors.py" - spec = importlib.util.spec_from_file_location("_real_errors_extraction_test", path) - module = importlib.util.module_from_spec(spec) - spec.loader.exec_module(module) - return module + """Import the real module under its own name, with conftest's Mock put back after. + + Loading it from a file path under a different module name also works, but coverage + does not credit the execution, so the extraction below would read as untested and + invite someone to delete it.""" + import sys + + saved = {name: sys.modules[name] for name in + ('python_modules.shared.errors', 'shared.errors') if name in sys.modules} + for name in saved: + del sys.modules[name] + try: + yield importlib.import_module('python_modules.shared.errors') + finally: + for name in list(sys.modules): + if name in ('python_modules.shared.errors', 'shared.errors'): + del sys.modules[name] + sys.modules.update(saved) DENIAL_V2 = """An error occurred while calling o304.load. @@ -76,6 +88,33 @@ def test_sdk_v1_pattern_still_works(self, errors): "Requested resource not found (Service: AmazonDynamoDBv2; Status Code: 400)") assert errors.get_error_message(Exception(v1)) == "Requested resource not found" + def test_spark_parse_exception_splits_off_the_sql(self, errors): + """Spark's ParseException carries the offending SQL after a marker line; the + function pulls it onto one line. Pre-existing behaviour, untested until this + file had a way to import the module.""" + class ParseError(Exception): + desc = "Syntax error at or near 'FRM'\n== SQL ==\nSELECT * FRM t\n ^^^" + + assert errors.get_error_message(ParseError()) == ( + "Syntax error at or near 'FRM' | SQL: SELECT * FRM t | ^^^" + ) + + def test_parse_exception_without_a_sql_section(self, errors): + class ParseError(Exception): + desc = " Syntax error, nothing more " + + assert errors.get_error_message(ParseError()) == "Syntax error, nothing more" + + def test_exception_with_a_message_attribute(self, errors): + """Some Py4J/Java wrappers expose .message rather than a useful str().""" + class WithMessage(Exception): + message = " the useful part " + + def __str__(self): + return "an unhelpful repr" + + assert errors.get_error_message(WithMessage()) == "the useful part" + def test_a_plain_exception_is_unchanged(self, errors): assert errors.get_error_message(ValueError("just a message")) == "just a message" diff --git a/tools/bulk_executor/tests/server/test_root.py b/tools/bulk_executor/tests/server/test_root.py index e309e399..3a102b06 100644 --- a/tools/bulk_executor/tests/server/test_root.py +++ b/tools/bulk_executor/tests/server/test_root.py @@ -630,6 +630,19 @@ def test_generic_exception_is_reported_then_exits(self, monkeypatch, capsys): "an unexpected failure prints its traceback where the user will see it" ) + def test_a_verbs_own_exit_passes_straight_through(self, monkeypatch, capsys): + """A helper that already called exit() has said its piece. Re-reporting it would + relabel a deliberate exit as a surprise and print a traceback for it.""" + run_mock = MagicMock(side_effect=SystemExit("PITR must be enabled first")) + verb = _make_verb_module("copy", run_callable=run_mock) + with pytest.raises(SystemExit) as exc_info: + _load_root(monkeypatch, + ["root.py", "--XAction", "copy"], + verb_module=verb, verb_name="copy") + + assert str(exc_info.value) == "PITR must be enabled first", "unchanged, not re-wrapped" + assert "did not expect" not in capsys.readouterr().out + def test_generic_exception_skips_commit_and_stop(self, monkeypatch): """A failed job must not be committed -- that is what makes Glue mark it FAILED.""" run_mock = MagicMock(side_effect=ValueError("nope")) From 949c7fcec798dfb6ebacc05408b46e0612c4e6d4 Mon Sep 17 00:00:00 2001 From: Jason Hunter Date: Tue, 1 Sep 2026 15:25:14 -0700 Subject: [PATCH 7/8] [bulk] Name the understood-failure marker so Spark's wording cannot trip it The client matches markers as substrings anywhere in a log line, so "Failure: " was too generic: Spark emits shapes like "ExecutorLostFailure: executor 7 exited" and "FetchFailure: shuffle block missing", either of which contains it. A false match suppresses Glue's diagnostics for a failure nobody explained -- the opposite of what the marker is for. Across every run captured for #332 the string appeared only in our own output, so this is about the shapes we have not seen rather than one we have. Now "Bulk Executor failure: ", with a guard test over four real Spark and Glue lines; reverting the marker fails it. Re-verified live: a denied count is 31 lines, no traceback or Glue blob, closing on AWS's own sentence. --- tools/bulk_executor/client/src/runner.py | 2 +- .../python_modules/shared/driver_errors.py | 8 ++++++- .../bulk_executor/tests/client/test_runner.py | 24 ++++++++++++++++++- 3 files changed, 31 insertions(+), 3 deletions(-) diff --git a/tools/bulk_executor/client/src/runner.py b/tools/bulk_executor/client/src/runner.py index ba0106f9..ad751c39 100644 --- a/tools/bulk_executor/client/src/runner.py +++ b/tools/bulk_executor/client/src/runner.py @@ -64,7 +64,7 @@ 'BulkExecutorError', 'A worker failed in a way we did not expect. Traceback from the worker:', 'The job failed in a way we did not expect. Traceback:', - 'Failure: ', + 'Bulk Executor failure: ', ) # Timing constants diff --git a/tools/bulk_executor/server/src/python_modules/shared/driver_errors.py b/tools/bulk_executor/server/src/python_modules/shared/driver_errors.py index 13866a0a..200c78a1 100644 --- a/tools/bulk_executor/server/src/python_modules/shared/driver_errors.py +++ b/tools/bulk_executor/server/src/python_modules/shared/driver_errors.py @@ -32,7 +32,13 @@ # suppress Glue's exception-analysis blob. Glue emits that blob for a clean sys.exit only # sometimes -- observed on a denied `find` but not a denied `count` in the same batch -- so # the marker has to be there every time, not only when a BulkExecutorError was involved. -EXPLAINED_FAILURE_PREFIX = "Failure: " +# +# Named rather than generic because the client matches it as a substring anywhere in a log +# line. "Failure: " alone appeared only in our own output across every run captured for +# #332, but Spark has shapes like ExecutorLostFailure and FetchFailure that could put +# "Failure: " in a line of its own -- and a false match would suppress Glue's diagnostics +# for a failure we had not explained. +EXPLAINED_FAILURE_PREFIX = "Bulk Executor failure: " # Cap on what we hand to sys.exit(): Glue records it as the job's ErrorMessage and the # client prints it as its closing line. AWS's authorization sentences run ~400 chars, diff --git a/tools/bulk_executor/tests/client/test_runner.py b/tools/bulk_executor/tests/client/test_runner.py index f1093ad1..9e47edeb 100644 --- a/tools/bulk_executor/tests/client/test_runner.py +++ b/tools/bulk_executor/tests/client/test_runner.py @@ -356,7 +356,7 @@ def test_driver_failure_marker_suppresses_the_glue_blob(self, bulk_runner, capsy monkeypatch.setattr(runner_module.utils, 'STD_ERROR_MESSAGE_KEYS', []) denial = _make_event(message=( - "Failure: User: arn:aws:sts::1:assumed-role/R/GlueJobRunnerSession is not " + "Bulk Executor failure: User: arn:aws:sts::1:assumed-role/R/GlueJobRunnerSession is not " "authorized to perform: dynamodb:Scan on resource: table/t" )) bulk_runner._pretty_print_log_event(denial) @@ -394,6 +394,28 @@ def test_our_own_sql_query_error_still_prints(self, bulk_runner, capsys): captured = capsys.readouterr() assert 'UNRESOLVED_COLUMN' in captured.out + captured.err + def test_sparks_own_failure_wording_does_not_trip_the_marker(self, bulk_runner, monkeypatch): + """The client matches markers as substrings, so a generic word would misfire. Spark + has plenty of "...Failure" shapes; none of them may be mistaken for the job saying + it has explained itself, or Glue's diagnostics get suppressed for a failure nobody + described.""" + monkeypatch.setattr(runner_module.utils, 'LOG_PATTERN_IGNORE_LIST', []) + monkeypatch.setattr(runner_module.utils, 'CONFIG_LOG_MESSAGE_KEYS', []) + monkeypatch.setattr(runner_module.utils, 'STD_ERROR_MESSAGE_KEYS', []) + + for spark_line in ( + "ERROR TaskSetManager: Lost task 3.0 in stage 2.0: ExecutorLostFailure: " + "executor 7 exited unrelated to the running tasks", + "WARN TaskSetManager: Lost task 1.0: FetchFailure: shuffle block missing", + 'ERROR GlueExceptionAnalysisListener:9 - {"Failure Reason": "boom"}', + "ERROR DAGScheduler: Job aborted due to stage failure: Task 0 failed 4 times", + ): + bulk_runner._suppress_glue_noise = False + bulk_runner._pretty_print_log_event(_make_event(message=spark_line)) + assert bulk_runner._suppress_glue_noise is False, ( + f"{spark_line[:60]!r} must not read as the job explaining itself" + ) + def test_real_worker_traceback_still_prints(self, bulk_runner, capsys): """Guard for #334: the anchor is Glue's reporter, not the frames. An unexpected worker failure prints its traceback through the same path and must survive.""" From c74a87293ce44ac6000761040517c74bfcc8a40e Mon Sep 17 00:00:00 2001 From: Jason Hunter Date: Tue, 1 Sep 2026 15:41:33 -0700 Subject: [PATCH 8/8] [bulk] Stop load printing an ERROR line on a run that then succeeds Reading zero items logged at ERROR and returned, so job.commit() ran, Glue marked the run SUCCEEDED, and the output read: ERROR - No data found, please check your data source Job completed successfully. Job duration: 0:01:28 An error and a success about the same run, with exit 0 for anything watching. Now a warning that says what happened and what to check, naming the path and the format the user claimed: "Read 0 items from '' as 'json' -- nothing was loaded. If that is unexpected, check --format and the path." Deliberately still a success. An empty drop is a legitimate input and the export pipeline already treats a 0-item export that way. Whether a source that holds bytes but yields no rows should fail instead -- the wrong-format case -- is a behaviour change with its own before/after, tracked as #340. --- .../src/python_modules/load/__init__.py | 9 ++++++- tools/bulk_executor/tests/server/test_load.py | 24 +++++++++++++++++++ 2 files changed, 32 insertions(+), 1 deletion(-) diff --git a/tools/bulk_executor/server/src/python_modules/load/__init__.py b/tools/bulk_executor/server/src/python_modules/load/__init__.py index 9e9d18d4..e82d42c7 100644 --- a/tools/bulk_executor/server/src/python_modules/load/__init__.py +++ b/tools/bulk_executor/server/src/python_modules/load/__init__.py @@ -91,7 +91,14 @@ def run(job, spark_context, glue_context, parsed_args): dynamicFrame = read_data(glue_context, s3_path, parsed_args) count = dynamicFrame.count() if count == 0: - log.error("No data found, please check your data source") # Should perhaps check that the path exists + # Logged as a warning, not an error: this run goes on to succeed, and an ERROR + # line above "Job completed successfully" is a contradiction the reader has to + # resolve. An empty drop is also a legitimate input -- the export pipeline + # treats a 0-item export the same way. Whether a source that holds bytes but + # yields no rows should fail instead is a separate question (#340). + log.warning( + f"Read 0 items from '{s3_path}' as {parsed_args.get('format')!r} -- " + f"nothing was loaded. If that is unexpected, check --format and the path.") return log.info(f"\nPreparing to load {count} items") log.info("Schema is:") diff --git a/tools/bulk_executor/tests/server/test_load.py b/tools/bulk_executor/tests/server/test_load.py index 5ba9318a..a08ba256 100644 --- a/tools/bulk_executor/tests/server/test_load.py +++ b/tools/bulk_executor/tests/server/test_load.py @@ -273,6 +273,30 @@ def test_returns_early_when_count_is_zero(self, monkeypatch): {'table': 't', 's3_path': 's3://b/k', 'format': 'csv'}) assert result is None + def test_zero_rows_warns_and_succeeds(self, monkeypatch, caplog): + """A run that goes on to succeed must not print an ERROR line. Measured before this + change: "ERROR - No data found, please check your data source" immediately above + "Job completed successfully", with exit 0.""" + import logging + + monkeypatch.setattr(load_module, 'check_s3_file_exists', lambda uri: True) + df = MagicMock() + df.count.return_value = 0 + monkeypatch.setattr(load_module, 'read_data', lambda *a: df) + write = MagicMock() + monkeypatch.setattr(load_module, 'write_dynamodb_dataframe', write) + + with caplog.at_level(logging.WARNING): + load_module.run(MagicMock(), MagicMock(), MagicMock(), + {'table': 't', 's3_path': 's3://b/k', 'format': 'json'}) + + levels = {r.levelname for r in caplog.records} + assert 'ERROR' not in levels, "a successful run must not log an error" + message = ' '.join(r.message for r in caplog.records) + assert 'Read 0 items' in message and "'json'" in message and '--format' in message, \ + "say what happened and what to check" + write.assert_not_called() + def test_raises_when_the_frame_cannot_even_be_created(self, monkeypatch): """Parquet reads its footer while the frame is created, so `--format parquet` at a CSV file raises from read_data rather than from count(). Both are the same mistake