TPC-H SF1000 Q17 stops completing after #2173; task-launch message overflow is misread as executor loss

[avantgardnerio]

TPC-H SF1000 Q17 stopped completing on main; bisected

Q17 completes in 77-134s on 696ca29b and does not complete at all on current
main (5e545034c). I killed it twice at 21 and 32 minutes. git bisect
across 7 steps names 9fecadae8 (#2173, "make gRPC max message size
consistently configurable") as the first bad commit.

Measured on a 32-executor EKS cluster reproducing the shape in
benchmarking.md: 4x r6i.24xlarge, 8 executors per node, 8 vCPU + 64 GiB each,
target_partitions=256, AQE on, dedicated 1000 GiB gp3 per executor, TPC-H
SF1000 ZSTD parquet on S3 in-region. DataFusion is 54.1.0 on both ends of the
range, so this is not a DataFusion bump.

Bisect

git bisect names 9fecadae8 (#2173) as the first bad commit, over 7 steps.

Predicate: does Q17 finish within 420s. Good samples ranged 77-134s, so there
is roughly 4x margin.

commit		Q17
696ca29b	GOOD	93.7s, and 133.6s on a repeat
828d6f007 #2186	GOOD	77.0s
187a65063 #2210 reqwest 0.12->0.13	GOOD	107.8s
234936485 #2208	GOOD	83.2s
63fa496b5 (CI-only diff)	GOOD	96.8s
a3161a301 #2212	GOOD	94.8s
9fecadae8 #2173	BAD	>420s
984513ee7 #2225	BAD	>480s
5e545034c main	BAD	>1200s

63fa496b5 changes only .github/workflows/codeql.yml, so its binaries are
c43e56eb4 (#2196) and that commit is cleared along with 6e9ff68a5 (#2215).

Two distinct failure signatures, not one

This is the part I would not have guessed, and it is why I am reporting rather
than proposing a fix.

On main, Q17 hangs in the scan/shuffle stages:

stage	status	tasks	elapsed
0	Running	512	2017s
1	Running	512	2017s
2	Successful	256	15.5s
3	Unresolved	0

Stage 0 is SortShuffleWriterExec: partitioning=Hash([l_partkey], 256) over all
6.00B rows of lineitem, with write_time=1077s and repart_time=75s, zero
spilling. Note that part filters to 199,303 rows out of 200M (0.1%
selectivity) and is correctly broadcast via HashJoinExec mode=CollectLeft,
so the plan choice looks right; lineitem is nonetheless fully hash-shuffled,
223.5 GB, on the probe side.

On 9fecadae8, stages 0-2 all complete quickly and correctly at 256 tasks
each (12.5s, 2.0s, 12.4s) and stage 3 balloons to 2168 tasks and never
finishes, with the scheduler logging:

Failed to launch new task: Failed to connect to executor <id>:
  Status { code: Internal, message: "h2 protocol error: http2 error",
  source: ... Reset(StreamId(67), INTERNAL_ERROR, User) }
-> Removing executor <id>

192 h2 protocol errors and 94 executor evictions in one run. The evicted
executors' subsequent status reports come back as Not executor with ID ... found.

So there appear to be at least two problems, and my 420s predicate cannot
distinguish them. Treat the bisect as locating the first commit where Q17 stops
completing, not as an explanation of main's stage-0 behaviour.

On the h2 resets

Reset(..., INTERNAL_ERROR, User) is what tonic emits when a message exceeds
the local limit, which points at #2173, where task-assignment clients gained:

let client = ExecutorGrpcClient::new(connection)
    .max_encoding_message_size(grpc_client_config.max_message_size)
    .max_decoding_message_size(grpc_client_config.max_message_size);

grpc_client_max_message_size defaults to 16 MiB, where previously tonic's own
defaults applied. But raising it to 134217728 on the scheduler did not stop
the resets: 192 errors and 94 evictions persisted with the flag verified present
on the pod. So either the message genuinely exceeds 128 MiB, or the flag is not
reaching the client that is failing.

Independently of the limit's value, a message-size overflow on task launch gets
classified as an unreachable executor, which triggers eviction plus unbounded
retry rather than surfacing the size problem. That is what turns it into a hang
instead of an error. To be fair to #2173, that classification looks like
pre-existing behaviour in launch_tasks / ExecutorManager, and #2173 only
supplies a new way to trigger it.

#2212 tested GOOD at 94.8s, so it is cleared and is not a contributing factor,
though it is worth noting that #2212 is the fix for #2029 ("Job hangs indefinitely instead of
failing when all executors are lost"). Its grace timer fails running jobs once
the cluster is empty, but here executors are evicted and then re-register, so
the cluster is never simultaneously empty, the timer never arms, and the job
spins on retries indefinitely.

Reproducing

--query 17 --path s3://<bucket>/tpch/sf1000/parquet --format parquet
--partitions 256 --iterations 1
-c datafusion.execution.collect_statistics=true
-c datafusion.execution.listing_table_factory_infer_partitions=false
-c ballista.planner.adaptive.enabled=true
-c ballista.shuffle.sort_based.memory_limit_per_task_bytes=0

Scheduler and executors with --grpc-server-max-{decoding,encoding}-message-size=134217728.
Whether --grpc-client-max-message-size is left at its 16 MiB default or set to
134217728 makes no difference to the outcome.

Related

• Encoded message too large #2165 (Encoded message too large) is the same underlying pressure on the
  client-submission path, where it surfaces as a visible error rather than a hang.
• Chunk oversized shuffle IPC messages so they never exceed the gRPC frame limit #2172 chunked oversized shuffle IPC messages; task assignment is a separate path.
• Job hangs indefinitely instead of failing when all executors are lost #2029 / fix(scheduler): fail running jobs when all executors are lost #2212 cover the executor-loss hang, but the grace timer does not fire
  here because executors re-register rather than staying lost.

Caveats

• Our Q17 on 696ca29b is 133.6s against the published 47.29s. Cluster shape is
  the likely cause: we pack 8 executors per r6i.24xlarge, and benchmarking.md
  does not state a node count, so per-pod memory and EBS bandwidth may differ
  substantially from the published run.
• The Dockerfiles at 696ca29b predate chore(docker): ca-certificates + h2o binary in benchmarks image #2251 and ship no ca-certificates, so
  s3:// reads fail with InvalidCertificate(UnknownIssuer). I backported that
  one RUN line for every commit in the range so it could not be mistaken for
  the regression.
• Q11 completes for us where the published results show FAIL, but returns 0 rows.
  Unverified; possibly the spec's 0.0001 / SF threshold being hardcoded.