RejectedExecutionException at StreamReader.read() submit strands Blocks; Block.awaitData() then blocks forever (unbounded wait regression)

### Analytics Accelerator Library for Amazon S3 Version

`1.3.1`

### AWS Region

Not region-specific. Reproduced against a local S3-compatible endpoint with no AWS account involved.

### Describe the running environment

Apache Spark Structured Streaming application (Spark `3.5.3`) reading Apache Iceberg tables, with
AAL enabled through Iceberg's `S3FileIO` (`s3.analytics-accelerator.enabled=true`).

The application runs **many concurrent streaming queries in one driver JVM**. Each query gets its own
Spark session, therefore its own `CatalogManager` → catalog → `S3FileIO`, and therefore its own AAL
`S3SeekableInputStreamFactory`. In the affected application that is several dozen queries across two
catalogs — comfortably more than 100 live `S3FileIO` instances in one JVM.

Full version set used for the report and the attached reproduction:

| Component | Version |
|---|---|
| `software.amazon.s3.analyticsaccelerator:analyticsaccelerator-s3` | `1.3.1` |
| `org.apache.iceberg:iceberg-aws`, `iceberg-core` | `1.10.1` (defect also present in `1.11.0` and `main`) |
| `org.apache.spark:spark-sql_2.12` | `3.5.3` |
| `software.amazon.awssdk` (s3, kms, sts, glue, dynamodb) | `2.29.52` (production path used BOM `2.42.13`) |
| `software.amazon.awssdk.crt:aws-crt` | `0.43.4` (reproduced with CRT **disabled**, see below) |
| `com.github.ben-manes.caffeine:caffeine` | `3.1.8` |
| `dev.failsafe:failsafe` | `3.3.2` |
| JDK | Corretto `17.0.15`, also `eclipse-temurin:17` |
| Kubernetes (for the containerised runs) | `kind` node image `kindest/node:v1.35.0` |
| S3 endpoint for reproduction | MinIO `RELEASE.2025-04-22T22-12-26Z` |

Every Iceberg and AAL setting other than `s3.analytics-accelerator.enabled=true` and
`s3.crt.enabled=false` is at its default — in particular AAL's `physicalio.thread.pool.size` (96),
`max.memory.limit` (2 GB) and `small.objects.prefetching.enabled` (`true`).

Note on CRT: this is **not** issue #263. That one requires the CRT client on aws-sdk-java-v2
`< 2.31.30`. Here the SDK is well above that threshold, and both reproductions hang with
`s3.crt.enabled=false` (Netty async client) — one of them uses no AWS SDK at all.

---

## What happened

**Summary.** A long-running Spark streaming application silently stopped doing work for several
hours. No exception, no failed task, no crash, no alert; application status still `RUNNING`. Roughly
three-quarters of its streaming queries had stopped part-way through starting up, and its data
sources fell hours behind. The cause is that reader threads are parked forever inside
`Block.awaitData()`: a `RejectedExecutionException` thrown *by* `threadPool.submit()` in
`StreamReader.read()` bypasses the recovery path that would have released them. Reproduced from
scratch with **AAL and three of its own runtime dependencies — no Iceberg, no AWS SDK, no network, no
credentials.**

### Preconditions

The trigger is that `S3SeekableInputStreamFactory.close()` — which ends in
`threadPool.shutdown()` — is called while streams created from that factory are still being read.

In our case the caller is Iceberg, which keeps factories in a `maximumSize(100)` Caffeine cache and
closes them from the cache's removal listener on *size eviction*, so a factory a reader still holds
gets closed underneath it. That is a caller-side bug and is filed separately as apache/iceberg#17485.

**This report is not about who called `close()`.** It is about what AAL does when it happens: the
rejection escapes the recovery path that already exists, and the reader then waits forever. And two of
the three paths below need **no caller mistake at all** — they are reachable with a perfectly
well-behaved caller.

### Symptom

Under concurrent startup, streaming-query threads **park permanently** inside `Block.awaitData()`.
They never return and never throw. Consequences, in the order an operator notices them:

1. The Spark application stays `RUNNING` and the driver stays healthy.
2. Affected queries are **absent from the Spark UI** — they hang before `query.start()` registers
them with the `StreamingQueryManager`, so they are not shown as failed; they are not shown at all.
3. No progress metrics, no exception, no crash, therefore no retry and no alert. Supervision code
that reacts to a thrown exception or a returned query observes neither.
4. Source lag grows silently.

### The sequence, in order

```text
Reader thread A F's BlockStore F's threadPool Any other thread
(holds factory F) (96 readers) (closes factory F)
| | | |
1 |------------------> | |
| makeRangeAvailable | |
| BlockManager:225-238 adds the Block to the store FIRST,
| latch = 1, and nothing owns it yet |
| | | |
2 | | <------------------|
| close() -> shutdown() | |
| the pool now rejects every submit |
| | | |
3 |-----------------------------------> |
| StreamReader.read() -> submit(task) |
4 <==== RejectedExecutionException ===| |
| thrown BY submit() at StreamReader:144, OUTSIDE the task, so
| the task's catch -- and the setErrorOnBlocksAndRemove()
| recovery it calls -- never run |
| | | |
| the Block is still present and still unfilled, and
| BlockStore:151-160 tests presence rather than readiness,
| so the range reads back as "available" and is never re-fetched
| | | |
5 |---> Block.awaitData() -> dataReadyLatch.await() |
| no timeout, and nothing left that can count it down
| >>>> PARKS FOREVER (Block:187) |
v
time
```

The **only** evidence visible from inside the application is AAL's own telemetry, one line per
*stranded* read — meaning a `Block` that has been registered in the `BlockStore` but never
filled, whose `dataReadyLatch` is never counted down, so any reader waiting on it waits forever:

```
[failure] [] block.manager.make.range.available(generation=0, thread_id=,
range=8000-15999, etag="", uri=s3:////metadata/.metadata.json,
range.effective=8000-73535): ns
[java.util.concurrent.RejectedExecutionException: 'Task FutureTask@… rejected from
java.util.concurrent.ThreadPoolExecutor@…[Shutting down, pool size = 1, active threads = 1,
queued tasks = 0, completed tasks = 0]']
```

Two details in that line matter: every failing read is on a small Iceberg `*.metadata.json` object,
and the rejecting pool frequently shows `completed tasks = 0` or `1` — it was created and destroyed
having done almost nothing.

In one incident this produced **tens of thousands of such lines over several hours**, while roughly
three-quarters of the application's streaming queries were silently absent. None before AAL was
enabled; none after it was disabled.

---

## Why it happens — exact lines at `v1.3.1`

### 1. The rejection is thrown *by* `submit()`, outside the task

`io/physical/reader/StreamReader.java:142-145`:

```java
public void read(@NonNull final List blocks, ReadMode readMode) {
Preconditions.checkArgument(!blocks.isEmpty(), "`blocks` list must not be empty");
threadPool.submit(processReadTask(blocks, readMode)); // <-- line 144
}
```

The recovery routine already exists and does exactly the right thing —
`setErrorOnBlocksAndRemove(blocks, error)` at `StreamReader.java:380-385` sets the error on every
not-ready `Block` (which counts its latch down via `Block.setError()`) and removes them from the
store. But it is only reachable from the `catch (Exception e)` at `StreamReader.java:216`, which is
**inside** the `Runnable` built by `processReadTask`. When `submit()` itself throws
`RejectedExecutionException`, that task never runs, so the recovery never runs.

### 2. The Blocks were already registered before the submit

`io/physical/data/BlockManager.java:225-238` adds every block in a group to the `BlockStore` and only
then calls `streamReader.read(blocksToFill, readMode)`. So at the moment of rejection the store
contains Blocks with `dataReadyLatch` at 1 and no owner.

### 3. Readiness is presence, not readiness

`io/physical/data/BlockStore.java:151-160`:

```java
private List getMissingBlockIndexesInRange(int startIndex, int endIndex) {
List missingBlockIndexes = new ArrayList<>();
for (int i = startIndex; i <= endIndex; i++) {
if (!blocks.containsKey(i)) {
missingBlockIndexes.add(i);
}
}
return missingBlockIndexes;
}
```

A Block that is present but never filled is indistinguishable from a ready one. Both callers depend
on this: `BlockManager.isRangeAvailable` (`:279-283`) reports the range available, and
`makeRangeAvailable` (`:152-207`) returns early. No later read will ever re-request it.

### 4. The wait for it is unbounded — and this is a regression

`io/physical/data/Block.java:185-197`:

```java
private void awaitData() throws IOException {
try {
dataReadyLatch.await(); // <-- line 187, no timeout
} catch (InterruptedException e) {
throw new IOException("Error while reading data. Read interrupted while waiting for data", e);
}
...
}
```

The latch is counted down in exactly two places, `setData()` (`:166`) and `setError()` (`:176`), and
both are reachable only from inside the task that never ran.

**This used to be bounded.** PR #340 ("Adapt retries to new PhysicalIO") replaced the timed wait:

```diff
private void awaitData() throws IOException {
try {
- if (!dataReadyLatch.await(readTimeout, TimeUnit.MILLISECONDS)) {
- throw new IOException(
- "Error while reading data. Request timed out after "
- + readTimeout
- + "ms while waiting for block data");
- }
+ dataReadyLatch.await();
} catch (InterruptedException e) {
```

The stated reasoning is sound on its own terms — *"we are merging timeout and retries into a unified
place … Failsafe now manages retries and timeouts"* — but the Failsafe timeout is applied to the
retry strategy **inside** the task. `StreamReader.createRetryStrategy()` (`:112-125`) builds it and
`:170` executes it, on the pool thread, measured from the moment the supplier begins running. It
cannot bound a consumer waiting for a task that never started.

Before #340 the three stranding paths below all terminated the reader after `blockreadtimeout` ms.
After #340 they hang forever.

### The small-object prefetch hides it

`BlockManager.prefetchSmallObject()` (`:122-129`) runs from the constructor for every object under
`physicalio small.object.size.threshold` (8 MB default, enabled by default) and catches `Exception`,
logging at `DEBUG` only, without removing the Blocks it already added. Iceberg metadata files are
always small, which is why every observed failure was on a `*.metadata.json`.

---

## Three ways to strand a Block, not one

| # | Path | Bounded before #340? | Bounded now? |
|---|---|---|---|
| 1 | `submit()` rejected — pool shut down | yes | **no** |
| 2 | Task throws an `Error` (`OutOfMemoryError`, `NoClassDefFoundError`); the handler is `catch (Exception e)` at `:216`, so the latch stays at 1 | yes | **no** |
| 3 | Task sits in the pool's unbounded `LinkedBlockingQueue` and never starts, so its Failsafe timer never starts | yes | **no** |

---

## Relation to #368

Issue #368 (open) reports `RejectedExecutionException` during concurrent S3 reads and is the same
*class* of problem — a thread pool shut down while AAL reads are in flight. It is worth reading the
two together, because the difference is instructive:

| | #368 | this report |
|---|---|---|
| Pool that shuts down | the AWS SDK's internal `ScheduledThreadPoolExecutor` | AAL's own `newFixedThreadPool` in `S3SeekableInputStreamFactory` |
| Where the rejection lands | **inside** `processReadTask` (`StreamReader.java:193` in that stack trace) | **at** `submit()` (`StreamReader.java:144`) |
| Recovery reachable? | yes — the task's `catch` runs | **no** |
| Observed behaviour | loud `IOException`, ~80% task failure | **silent permanent hang** |

So #368 is independent evidence that the in-task error path works as designed. The defect here is
the same exception arriving two frames earlier, where that path is unreachable. Fixing the submit
site would also make #368's failure mode strictly better-behaved.

---

## Reproduction

The complete reproduction is inline below — self-contained, **no AWS credentials or account
required**. The containerised and Spark harnesses, plus the candidate patches, are omitted here for
length and I am happy to share them or open a PR.

### A. AAL only — four jars, no network, no credentials (~45 s)

The reproduction below uses **only
`analyticsaccelerator-s3` and three of its own runtime dependencies**. AAL 1.3.1 declares no
transitive dependencies, so the compile classpath is a single jar; `slf4j-api`, an slf4j binding and
`failsafe` are needed at runtime. No Iceberg, no Caffeine, no AWS SDK, no reflection, no network.

```bash
AAL=…/analyticsaccelerator-s3-1.3.1.jar
javac -cp "$AAL" -d out src/software/amazon/s3/analyticsaccelerator/AalMre.java
java -cp "out:$AAL:$SLF4J_API:$SLF4J_SIMPLE:$FAILSAFE" \
software.amazon.s3.analyticsaccelerator.AalMre broken
```

It builds a factory over an in-memory `ObjectClient`, disables nothing, and runs two scenarios:
**A** closes the whole factory then reads through it; **B** shuts down only the reader pool, leaving
the stores and telemetry live — which is the state your telemetry reports in the wild. All four reads
hang:

```
SCENARIO A factory.close() was called, then its holder reads through it
factory.close() done; pool.isShutdown=true
[A1] createStream(...) RESULT: returned normally
[A2] first read RESULT: HUNG (>15s). state=WAITING
at java.util.concurrent.CountDownLatch.await(CountDownLatch.java:230)
at …io.physical.data.Block.awaitData(Block.java:187)
[A3] second read RESULT: HUNG (>15s) (identical stack)
SCENARIO B only the reader pool is shut down; stores and telemetry still live
[B1] first read RESULT: HUNG (>15s)
[B2] second read RESULT: HUNG (>15s)
ALL EXPECTATIONS MET for mode=broken
```

With logging on, the same run prints the rejection and its full call chain, which is the mechanism in
one stack trace — note that it originates in the constructor-time small-object prefetch:

```
[failure] block.manager.make.range.available(…): ns
[java.util.concurrent.RejectedExecutionException: … rejected from
ThreadPoolExecutor@…[Terminated, pool size = 0, active threads = 0, completed tasks = 0]]
at java.util.concurrent.AbstractExecutorService.submit(AbstractExecutorService.java:123)
at …io.physical.reader.StreamReader.read(StreamReader.java:144)
at …io.physical.data.BlockManager.lambda$makeRangeAvailable$1(BlockManager.java:238)
at …io.physical.data.BlockManager.makeRangeAvailable(BlockManager.java:210)
at …io.physical.data.BlockManager.prefetchSmallObject(BlockManager.java:125)
at …io.physical.data.BlockManager.(BlockManager.java:115)
```

The harness takes `broken` or `fixed` and **exits non-zero on unmet expectations**, so it is a
regression test in both directions rather than a demonstration.

### A2. The caller-side half, if you want to see it

A companion reproduction drives Iceberg's real static factory cache by reflection and
shows the eviction closing a factory the caller still holds. It is **not** needed to reproduce this
issue and is only included so the whole picture is available; it is the subject of the companion
Iceberg report.

### B. Containerised, against a real S3 endpoint

130 concurrent readers against MinIO in a `kind` cluster. Manifests, Dockerfiles and run scripts are
all in `repro/`.

---

## Reproduction matrix

Readers are held constant at 130 in every row; the only variable is the number of distinct
`S3FileIO` instances, which is the number of distinct factory-cache keys. Holding the reader count
fixed keeps thread count, socket count and endpoint load identical, so the key count is isolated.

| AAL | Readers | Distinct factories | OK | Threw | **Hung** | Note |
|---|---|---|---|---|---|---|
| disabled | 130 | 130 | 130 | 0 | 0 | environment is healthy |
| enabled | 130 | 99 | 130 | 0 | 0 | under the cache bound |
| enabled | 130 | **130** | 100 | 20 | **10** | over the bound |
| enabled | 130 | 1 | 130 | 0 | 0 | single shared factory |
| enabled, `small.objects.prefetching.enabled=false` | 130 | 130 | 100 | 0 | **30** | **worse** — see below |
| enabled, with the patches below | 130 | 130 | **130** | 0 | **0** | |

Two observations from that table:

- In the broken row **exactly 100 readers succeed**, which is the bound of the cache holding the
factories.
- Disabling small-object prefetching does **not** help. The totals look similar but the composition
is materially worse: 0 thrown and 30 hung, versus 20 thrown and 10 hung with prefetching on. It
converts loud failures into silent ones. Raw pod log for that run: 30
`[failure] block.manager.make.range.available` lines in a 2,151-line capture.

---

## Proposed fix

Both patches were applied to the `v1.3.1` sources, compiled, and loaded ahead of the released jar by
classpath precedence. Full patched sources available on request, or as a PR if you prefer.

### Patch 1 (primary) — make the rejection reach the recovery that already exists

`io/physical/reader/StreamReader.java`:

```diff
public void read(@NonNull final List blocks, ReadMode readMode) {
Preconditions.checkArgument(!blocks.isEmpty(), "`blocks` list must not be empty");
- threadPool.submit(processReadTask(blocks, readMode));
+ try {
+ threadPool.submit(processReadTask(blocks, readMode));
+ } catch (RejectedExecutionException e) {
+ // The task never runs, so its own catch block cannot fail the blocks. Waiters on these
+ // blocks would otherwise block forever on a latch that is never counted down.
+ LOG.error("Read task rejected by the thread pool, failing blocks", e);
+ setErrorOnBlocksAndRemove(blocks, new IOException("Read task rejected by thread pool", e));
+ throw e;
+ }
}
```

**The same defect exists at a second call site**, and a patch that misses it leaves the bug live:
`io/physical/impl/PhysicalIOImpl.readVectored()` (`:270`) also does a bare `threadPool.submit(...)`,
in a loop, with the same SpotBugs justification *"we do not have any use for this Future"*. Its
pre-submit state is per-range: an allocated `ByteBuffer` and the caller-visible
`objectRange.getByteBuffer()` future. On rejection it should complete the rejected range
exceptionally, release that range's buffer, and fail every not-yet-submitted range. The attached
patch adds a small `failPendingRanges` helper for that. This matters increasingly, since the vectored
path is being adopted by callers.

Consider extracting a shared helper rather than duplicating the `try/catch`.

*Note on contract:* `read()` is documented as fire-and-forget, so rethrowing changes its contract.
`prefetchSmallObject` already catches `Exception`, but every call site should be audited rather than
assumed.

### Patch 2 (defence in depth) — restore a bounded wait

`io/physical/data/Block.java`:

```diff
private void awaitData() throws IOException {
try {
- dataReadyLatch.await();
+ if (!dataReadyLatch.await(blockReadDeadlineMillis, TimeUnit.MILLISECONDS)) {
+ throw new IOException("timed out waiting for block " + blockKey
+ + " after " + blockReadDeadlineMillis + " ms");
+ }
} catch (InterruptedException e) {
```

Patch 1 closes stranding path 1. Paths 2 and 3 remain reachable without it, and an unbounded wait on
a latch that only a remote party can release is a liveness hazard regardless of which bug reaches it.

The deadline is genuinely awkward and I do not claim it is solved: it must exceed the in-task budget
of `blockreadtimeout × (blockreadretrycount + 1)` = 30 s × 21 = **630 s** by default, or healthy slow
reads will be killed. A 10-minute backstop still beats infinity, but choosing something shorter means
revisiting the retry defaults too. My patch sources the deadline from a static default plus a system-property
override to avoid changing `Block`'s constructor signature.

### Verification

| Run | Classpath | Expectation | Result | Exit |
|---|---|---|---|---|
| Reproduction A | released jar | `broken` — must hang | all four reads HUNG | 0 ✓ |
| Reproduction A | + patches | `fixed` — must not hang | all four THREW `RejectedExecutionException` | 0 ✓ |
| Reproduction A | + patches | `broken` — *must fail* | correctly reported unmet expectations | 1 ✓ |
| Reproduction B, 130 readers / 130 factories | released jar | — | 100 ok / 20 threw / **10 hung** | — |
| Reproduction B, 130 readers / 130 factories | + patches | — | **130 ok / 0 / 0** in 2.46 s | — |

The third row is the one that makes the second meaningful: running the patched build against the
"must reproduce" expectation fails, so the passing result is not vacuous.

### A change I would *not* recommend

Making `BlockStore.getMissingBlockIndexesInRange` treat a present-but-unfilled Block as missing looks
like the direct fix, but a legitimately in-flight Block is also present-and-not-ready. Treating it as
missing means every concurrent reader re-issues the same range request — a thundering herd against
exactly the hot objects AAL exists to accelerate, plus duplicate Blocks in the store. Distinguishing
*in-flight* from *abandoned* is what Patch 1 does, at the point where abandonment happens.

---

## Smaller findings in the same area

1. **`close()` ordering.** `S3SeekableInputStreamFactory.close()` (`:193-199`) does
`objectMetadataStore.close(); objectBlobStore.close(); telemetry.close(); threadPool.shutdown();`.
The stores and telemetry are closed *before* the pool, so a task still running can touch an
already-closed `BlobStore`/`Telemetry`. There is no `awaitTermination`, no `shutdownNow`, no
drain, and no cancellation of submitted work. Note the inconsistency: `BlobStore.close()` uses
`shutdownNow()` for its maintenance scheduler while the pool doing the actual S3 work gets plain
`shutdown()`.
2. **`prefetchSmallObject` swallows at DEBUG** (`BlockManager.java:122-129`) and removes nothing.
Because `StreamReader.read` is fire-and-forget, the only exceptions this catch can observe are
the synchronous ones thrown before or between submits — i.e. precisely the ones that strand
Blocks. `RejectedExecutionException` became catchable here when the catch was widened from
`IOException` to `Exception`.
3. **Documentation.** The only published statement about factory lifetime is in `README.md`: *"When
the `S3SeekableInputStreamFactory` is no longer required to create new streams, close it to free
resources (eg: caches for prefetched data) held by the factory."* That precondition is
**satisfiable while streams created from the factory are still being read**, so a caller can
comply with it and still trigger this bug. It should state the second precondition explicitly —
that every stream already created must also be finished. Relatedly, the class javadoc's ownership
paragraph covers only the passed `ObjectClient`, not the factory's own thread pool and stores, and
it was written when the class was not `AutoCloseable` at all.
4. **`physicalio thread.pool.size` is undocumented.** It defaults to 96 threads per factory (plus one
`BlobStore` maintenance thread), and does not appear in `doc/CONFIGURATION.md`, while
`max.memory.limit` (2 GB, also per factory) is documented and even carries a sizing caveat. A
caller that creates many factories has no documented way to discover the per-factory cost.

---

## Expected behavior

A read that cannot be scheduled should **fail**, not hang. Concretely:

1. When `threadPool.submit()` is rejected, the affected `Block`s should be failed and removed so that
waiters receive an `IOException` rather than blocking on a latch nobody will count down.
2. `Block.awaitData()` should be bounded, so that any lost completion — not just this one — surfaces
as an error within a known time.
3. Closing a factory while streams created from it are still being read should either be prevented or
be safe; today it is neither, and the only published guidance on when it is safe to close
(`README.md`) does not mention in-flight streams.

## Relevant log output

```
[failure] [] block.manager.make.range.available(generation=0, thread_id=,
range=8000-15999, etag="", uri=s3:////metadata/.metadata.json,
range.effective=8000-73535): ns
[java.util.concurrent.RejectedExecutionException: 'Task java.util.concurrent.FutureTask@…
[Not completed, task = …[Wrapped task = software.amazon.s3.analyticsaccelerator.io.physical
.reader.StreamReader$$Lambda$…]] rejected from java.util.concurrent.ThreadPoolExecutor@…
[Shutting down, pool size = 1, active threads = 1, queued tasks = 0, completed tasks = 0]']
```

Thread dump of a reader that will never return:

```
java.util.concurrent.CountDownLatch.await(CountDownLatch.java:230)
software.amazon.s3.analyticsaccelerator.io.physical.data.Block.awaitData(Block.java:187)
software.amazon.s3.analyticsaccelerator.io.physical.data.Block.read(Block.java:125)
software.amazon.s3.analyticsaccelerator.io.physical.data.Blob.read(Blob.java:164)
software.amazon.s3.analyticsaccelerator.io.physical.impl.PhysicalIOImpl.read(PhysicalIOImpl.java:159)
software.amazon.s3.analyticsaccelerator.io.logical.impl.SequentialLogicalIOImpl.read(SequentialLogicalIOImpl.java:65)
software.amazon.s3.analyticsaccelerator.S3SeekableInputStream.read(S3SeekableInputStream.java:151)
```

## Workarounds for anyone hitting this now

Ordered by whether they actually prevent it.

| Workaround | Effect |
|---|---|
| **Disable AAL** (`s3.analytics-accelerator.enabled=false` on the catalog) | Prevents. The only complete mitigation without patching. |
| **Ensure the caller holds only one `S3SeekableInputStreamFactory` per JVM** — for Iceberg users, one shared `S3FileIO` per JVM (see the companion Iceberg issue) | Prevents in practice: with one cache key the factory is never evicted. Measured clean at 130 concurrent readers, and faster than the AAL-disabled baseline. |
| **Keep concurrent factory count comfortably below the caller's cache bound** | Reduces probability only. Establishes no invariant and does not survive scaling up. |
| Disable small-object prefetching | **Do not use.** Measured worse: converts thrown errors into silent hangs. |
| Tune `physicalio.thread.pool.size` or `max.memory.limit` | No effect on this bug. They change per-factory cost, not the close-while-in-use behaviour. |
| Add a startup/read timeout in the calling application | Detection, not prevention — and if the caller restarts into the same condition it becomes a crash loop. Note the parked thread waits on an uninterruptible latch and is never reclaimed. |

**Detection**, since the failure is silent: alert on the AAL telemetry failure line
(`block.manager.make.range.available` + `failure`), which has a zero baseline in a healthy system,
and monitor threads parked in `Block.awaitData` if you can. Do not rely on application-level health
signals — they are emitted by the component that hung.

---

## Code of Conduct

- [x] I agree to follow this project's Code of Conduct

---

**AI Disclosure**
- Model: Claude Opus 4.6
- Platform/Tool: Claude Code
- Human Oversight: fully reviewed
- Prompt Summary: Investigate a production incident in which streaming queries silently failed to
start with the Analytics Accelerator enabled; identify root cause from pinned sources, build
runnable reproductions, verify candidate patches, and draft an upstream bug report.

---

Full reproduction source — AalMre.java, AAL only, no other library on the classpath

```java
package software.amazon.s3.analyticsaccelerator;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import software.amazon.s3.analyticsaccelerator.request.GetRequest;
import software.amazon.s3.analyticsaccelerator.request.HeadRequest;
import software.amazon.s3.analyticsaccelerator.request.ObjectClient;
import software.amazon.s3.analyticsaccelerator.request.ObjectContent;
import software.amazon.s3.analyticsaccelerator.request.ObjectMetadata;
import software.amazon.s3.analyticsaccelerator.util.OpenStreamInformation;
import software.amazon.s3.analyticsaccelerator.util.S3URI;

/**
* Minimal reproduction of the AAL defect, using ONLY analyticsaccelerator-s3 and its own
* dependencies. No Iceberg, no Caffeine, no reflection, no AWS SDK, no network, no credentials.
*
*

Lives in AAL's own package so it can reach the package-private {@code getThreadPool()} accessor
* on {@link S3SeekableInputStreamFactory}. Nothing else here is privileged.
*
*

What it shows: once a factory's reader pool has been shut down, a read through that factory
* never completes and never throws. {@code StreamReader.read()} submits the fetch task with a bare
* {@code threadPool.submit(...)}, so a {@code RejectedExecutionException} is thrown by the submit
* itself, outside the task — which means the task's own catch, and the
* {@code setErrorOnBlocksAndRemove()} recovery it calls, never run. The {@code Block} that
* {@code BlockManager} already registered is left with its latch at 1, and
* {@code Block.awaitData()} waits on that latch with no timeout.
*
*

Two scenarios, because both orderings occur in practice:
*
*

*
• A — the whole factory was closed via {@code close()}. That is what a caller's cleanup
  * or a caller's cache eviction does.
  *
• B — only the reader pool has been shut down, while the factory's stores and telemetry
  * are still live. This is the state AAL's own telemetry reports in the wild (a
  * {@code ThreadPoolExecutor[Shutting down, …]} rejection recorded by a telemetry sink that is
  * evidently still working), and it is reachable inside {@code close()} itself, which shuts the
  * pool down after closing the stores.
  *

*
*

Usage: {@code AalMre [broken|fixed]}. {@code broken} asserts the defect is present;
* {@code fixed} asserts that reads fail fast instead of hanging. Exits non-zero on unmet
* expectations, so it serves as a regression test in both directions.
*/
public final class AalMre {

private static final int OBJECT_LEN = 128 * 1024;
private static final long WATCHDOG_MS = 15_000;

private static String mode = "broken";
private static final java.util.List FAILURES = new java.util.ArrayList<>();

private static boolean fixed() {
return "fixed".equals(mode);
}

public static void main(String[] args) throws Exception {
if (args.length > 0) {
mode = args[0];
}
System.out.println("analyticsaccelerator-s3 reproduction; expectation mode = " + mode);

scenarioA();
scenarioB();

banner("SUMMARY");
if (FAILURES.isEmpty()) {
System.out.println(" ALL EXPECTATIONS MET for mode=" + mode);
Runtime.getRuntime().halt(0);
}
System.out.println(" UNMET EXPECTATIONS for mode=" + mode + ":");
FAILURES.forEach(f -> System.out.println(" - " + f));
Runtime.getRuntime().halt(1);
}

/** The factory is fully closed, then used by a caller that still holds a reference to it. */
private static void scenarioA() throws Exception {
banner("SCENARIO A factory.close() was called, then its holder reads through it");

S3SeekableInputStreamFactory factory = newFactory();
factory.close();
System.out.println(
" factory.close() done; pool.isShutdown=" + factory.getThreadPool().isShutdown());

S3SeekableInputStream[] held = new S3SeekableInputStream[1];
System.out.println(" [A1] createStream(...)");
watched("A1-createStream", () -> held[0] = factory.createStream(S3URI.of("bucket", "k/1.json")));

if (held[0] == null) {
FAILURES.add("A1 createStream did not return a stream");
return;
}
System.out.println(" [A2] first read");
expect("A2 first read", watched("A2-read1", () -> held[0].read(new byte[64], 0, 64)));
System.out.println(" [A3] second read of the same range");
expect("A3 second read", watched("A3-read2", () -> {
held[0].seek(0);
held[0].read(new byte[64], 0, 64);
}));
}

/** Only the reader pool is shut down; the factory's stores and telemetry are still live. */
private static void scenarioB() throws Exception {
banner("SCENARIO B only the reader pool is shut down; stores and telemetry still live");

S3SeekableInputStreamFactory factory = newFactory();
S3SeekableInputStream stream = factory.createStream(S3URI.of("bucket", "k/1.json"));
factory.getThreadPool().shutdown(); // the last statement of close(), reached first by a racer
System.out.println(" pool.isShutdown=" + factory.getThreadPool().isShutdown());

System.out.println(" [B1] first read");
expect("B1 first read", watched("B1-read1", () -> stream.read(new byte[64], 0, 64)));
System.out.println(" [B2] second read of the same range");
expect("B2 second read", watched("B2-read2", () -> {
stream.seek(0);
stream.read(new byte[64], 0, 64);
}));
}

// ---------------------------------------------------------------------------------------------

private static S3SeekableInputStreamFactory newFactory() {
return new S3SeekableInputStreamFactory(
new FakeObjectClient(), S3SeekableInputStreamConfiguration.DEFAULT);
}

/** In "broken" mode a read must hang; once patched it must fail fast instead. */
private static void expect(String label, String outcome) {
boolean ok = fixed() ? !"HUNG".equals(outcome) : "HUNG".equals(outcome);
System.out.println((ok ? " [PASS] " : " [FAIL] ") + label + " -> " + outcome
+ " (expected " + (fixed() ? "not HUNG" : "HUNG") + ")");
if (!ok) {
FAILURES.add(label + " was " + outcome);
}
}

/** Runs {@code body} on a watched thread; returns HUNG, THREW or RETURNED. */
private static String watched(String label, ThrowingRunnable body) throws Exception {
final Throwable[] thrown = new Throwable[1];
final boolean[] returned = {false};
Thread t = new Thread(() -> {
try {
body.run();
returned[0] = true;
} catch (Throwable e) {
thrown[0] = e;
}
}, label);
t.setDaemon(true);
t.start();
t.join(WATCHDOG_MS);

if (t.isAlive()) {
System.out.println(" RESULT: HUNG (>" + WATCHDOG_MS / 1000 + "s). state=" + t.getState());
for (StackTraceElement e : t.getStackTrace()) {
String s = e.toString();
if (s.contains("analyticsaccelerator") || s.contains("CountDownLatch")
|| s.contains("AbstractQueuedSynchronizer") || s.contains("LockSupport")) {
System.out.println(" at " + s);
}
}
return "HUNG";
}
if (thrown[0] != null) {
Throwable root = thrown[0];
while (root.getCause() != null) {
root = root.getCause();
}
System.out.println(" RESULT: THREW " + thrown[0].getClass().getSimpleName()
+ " root=" + root.getClass().getName());
return "THREW";
}
System.out.println(" RESULT: returned normally");
return "RETURNED";
}

interface ThrowingRunnable {
void run() throws Exception;
}

/** In-memory stand-in for S3. No network, no credentials, no AWS SDK. */
private static final class FakeObjectClient implements ObjectClient {
@Override
public ObjectMetadata headObject(HeadRequest r, OpenStreamInformation i) {
return ObjectMetadata.builder().contentLength(OBJECT_LEN).etag("etag-1").build();
}

@Override
public ObjectContent getObject(GetRequest r, OpenStreamInformation i) {
return ObjectContent.builder().stream(new ByteArrayInputStream(new byte[OBJECT_LEN])).build();
}

@Override
public void close() throws IOException {}
}

private static void banner(String s) {
System.out.println();
System.out.println("================================================================");
System.out.println(s);
System.out.println("================================================================");
}
}
```

Start with io/physical/reader/StreamReader.java, especially read(), processReadTask(), and setErrorOnBlocksAndRemove(), then trace registration in BlockManager.java and readiness checks in BlockStore.java. Read Block.java's awaitData(), setData(), and setError() to understand the wait lifecycle. Done means rejected or otherwise stranded reads terminate and release their blocks instead of waiting indefinitely; use the issue's listed line ranges as the starting points for regression coverage.

Written by the indexing model from the issue text.