feat: configurable timeouts and concurrency with lenient defaults for high-burst workloads (#14)

## Summary

- Expose `connectionTimeout`, `connectionAcquireTimeout`,
`subscriptionTimeout`, and `maxConcurrency` on
`SageMakerConfig.Builder`. Each is validated `> 0` / non-null at build
time. Public `DEFAULT_*` constants on `SageMakerConfig` document the
chosen defaults.
- Wire them through `SageMakerTransportFactory` (Netty
`connectionTimeout` / `connectionAcquisitionTimeout` / `maxConcurrency`)
and `SageMakerTransport` (input-publisher subscription wait at
`ensureConnected()`).
- Move defaults to values tuned for high-burst workloads instead of
inheriting the AWS Netty defaults, which fail-fast in a way that's
actively wrong for SageMaker bidi streaming under bursts.

| Knob | Old default | New default | Why |
|---|---|---|---|
| `connectionTimeout` | AWS Netty default = 2 s | **30 s** | Cold-start
endpoints under a 200–500-stream burst can't accept TLS handshakes in 2
s |
| `connectionAcquireTimeout` | AWS Netty default = 10 s | **60 s** | A
400-stream burst drains the acquire pool past 10 s |
| `subscriptionTimeout` | (was hardcoded 30 s) | **60 s** | Slow
first-stream subscribe under burst |
| `maxConcurrency` | (was hardcoded 500) | **500** (now tunable) | No
change — just made it tunable |

## Why this matters

Empirically validated against a 400-concurrent-stream burst test on a
10× `ml.g6.2xlarge` endpoint, replicating a customer report:

| Error category | Before (Maven Central 0.1.2 + AWS defaults) | After
(this PR) |
|---|---:|---:|
| `Connection pool exhausted (Acquire took longer)` | 3,230 | ~0 |
| `TCP connect timed out` | 706 | ~0 |
| `TCP Connection reset` | 3,528 | sharply down |

Out of the box, callers no longer have to know that AWS's
general-purpose Netty defaults bite hard on the SageMaker streaming
path; the transport ships sane-for-its-use-case defaults. Anyone who
wants fail-fast behavior can tighten the knobs explicitly:

```java
SageMakerConfig config = SageMakerConfig.builder()
    .endpointName("my-deepgram-endpoint")
    .region("us-east-2")
    .connectionTimeout(Duration.ofSeconds(5))
    .connectionAcquireTimeout(Duration.ofSeconds(15))
    .build();
```

## Scope and what this does NOT fix

This PR addresses the transport layer only. The SDK-level retry storm in
`com.deepgram:deepgram-java-sdk` (the hardcoded
`connectionFuture.get(4000, MILLISECONDS)` in
`ReconnectingWebSocketListener.connect()` plus the listener's default
`maxRetries = Integer.MAX_VALUE`) is not touched here and was
empirically observed to remain even with this PR's transport fixes — the
bottleneck simply moves inward to the SDK ceiling.

The full burst-handling fix requires pairing this PR with the
corresponding change in `deepgram-java-sdk` that auto-disables
`ReconnectingWebSocketListener` when a custom transport is configured
(`ClientOptions.reconnect(false)` + auto-set in
`DeepgramClientBuilder.transportFactory(...)`). With both layers in
place, end-to-end error counts drop ~97 % at 400 concurrent streams.

## Backwards compatibility

Adds new builder methods only; no existing API is removed or modified.
Existing callers using
`SageMakerConfig.builder().endpointName(...).region(...).build()` get
the new defaults automatically — same calling convention, more lenient
runtime behavior. No code changes required for existing applications.

## Test plan

- [x] `./gradlew :sagemaker-transport:test` — all tests pass (18/18; 4
new tests for defaults, custom values, and validation of non-positive
arguments).
- [x] Local build of the patched JAR consumed by a load-test harness
against a real SageMaker endpoint.
- [x] Validation run vs. baseline measured a `Connection pool exhausted`
/ `TCP connect timed out` reduction to ~0.
- [ ] CI passes.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Co-authored-by: Greg Holmes <greg.holmes@deepgram.com>

Author: 3 people

README.md

@@ -11,7 +11,7 @@ SageMaker transport for the [Deepgram Java SDK](https://github.com/deepgram/deep
 
 ```groovy
 dependencies {
-    implementation 'com.deepgram:deepgram-java-sdk:0.2.1'
+    implementation 'com.deepgram:deepgram-java-sdk:0.4.0'
     implementation 'com.deepgram:deepgram-sagemaker:0.1.2' // x-release-please-version
 }
 ```
@@ -29,7 +29,7 @@ dependencies {
 ## Requirements
 
 - Java 11+
-- [Deepgram Java SDK](https://github.com/deepgram/deepgram-java-sdk) v0.2.1+
+- [Deepgram Java SDK](https://github.com/deepgram/deepgram-java-sdk) v0.4.0+ (the `default ReconnectOptions reconnectOptions()` hook on `DeepgramTransportFactory` is required for storm absorption)
 - AWS credentials configured (environment variables, shared credentials file, or IAM role)
 - A Deepgram model deployed to an AWS SageMaker endpoint
 
@@ -95,6 +95,15 @@ The transport is transparent — the SDK API is identical whether using Deepgram
 |-----------|----------|---------|-------------|
 | `endpointName` | Yes | — | SageMaker endpoint name |
 | `region` | No | `us-west-2` | AWS region |
+| `connectionTimeout` | No | `30s` | Max time for the underlying TCP/TLS connect (AWS Netty default is 2 s — bumped here so cold-start endpoints under burst load have time to accept TLS handshakes). |
+| `connectionAcquireTimeout` | No | `60s` | Max time to acquire a connection from the Netty pool (AWS Netty default is 10 s — bumped so a 200–500-stream burst doesn't drain the acquire pool). |
+| `subscriptionTimeout` | No | `60s` | Max time the transport waits for the AWS SDK to subscribe to the bidi-stream input publisher before failing. A timeout here is treated as a transient connect failure and counts against `maxRetries` / `retryBudget`. |
+| `maxConcurrency` | No | `500` | Max simultaneous in-flight HTTP/2 streams across the shared Netty pool. With `maxStreams=1` this is the cap on simultaneous bidirectional streams. |
+| `maxRetries` | No | `5` | Max retries on transient AWS errors (throttling, pool-exhausted, transient connect/timeout). Set to `0` to disable internal retry. Terminal errors (auth, validation) bypass this. |
+| `initialBackoff` | No | `100ms` | First backoff delay applied after the initial failure. |
+| `maxBackoff` | No | `5s` | Cap on the per-attempt backoff delay regardless of multiplier. |
+| `backoffMultiplier` | No | `2.0` | Exponential growth factor between retry attempts. Must be `>= 1.0`. |
+| `retryBudget` | No | `30s` | Total wall-clock cap across all retry attempts before giving up and surfacing the error to listeners. |
 
 ```java
 SageMakerConfig config = SageMakerConfig.builder()
@@ -103,6 +112,88 @@ SageMakerConfig config = SageMakerConfig.builder()
     .build();
 ```
 
+#### High-concurrency notes
+
+The transport's defaults are tuned for high-burst workloads (large numbers of
+streams opened in a tight loop against an endpoint that may need to scale up).
+If you're opening 200–500 streams simultaneously against a cold endpoint,
+the AWS Netty defaults (2 s connect / 10 s acquire) will fire before
+the load balancer has accepted all of the inbound TLS handshakes — you'll
+see a wave of `connection acquire` and `connect timed out` errors that look
+like server-side problems but are really client-side fail-fast tripping early.
+
+This transport ships with more lenient defaults (30 s / 60 s) so the
+common high-concurrency path works out of the box. Tighten them if you need
+fail-fast behavior in low-latency pipelines:
+
+```java
+SageMakerConfig config = SageMakerConfig.builder()
+    .endpointName("my-deepgram-endpoint")
+    .region("us-east-2")
+    .connectionTimeout(Duration.ofSeconds(5))
+    .connectionAcquireTimeout(Duration.ofSeconds(15))
+    .build();
+```
+
+#### Retry & storm absorption
+
+Transient AWS-side failures (`ThrottlingException`, connection-pool exhaustion, transient
+connect/timeout failures) are absorbed by the transport itself: classified as retryable, retried
+with exponential backoff up to `maxRetries` and `retryBudget`, with messages enqueued during the
+reset window persisted across the reconnect so audio isn't dropped. Only **terminal** errors (auth,
+validation) and budget-exhausted retryable errors propagate to `transport.onError(...)` and reach
+the application's error handler.
+
+This means the SDK's wrapper-level reconnect (`ReconnectingWebSocketListener`) would compound the
+plugin's internal retries into a Throttling-on-Throttling storm under burst load, so the plugin
+declares `ReconnectOptions.builder().maxRetries(0).build()` via the
+`DeepgramTransportFactory.reconnectOptions()` hook. The SDK applies it automatically when it sees
+a `transportFactory` in use; no user wiring required.
+
+To tune retry behavior:
+
+```java
+SageMakerConfig config = SageMakerConfig.builder()
+    .endpointName("my-deepgram-endpoint")
+    .maxRetries(10)
+    .initialBackoff(Duration.ofMillis(200))
+    .maxBackoff(Duration.ofSeconds(10))
+    .retryBudget(Duration.ofMinutes(1))
+    .build();
+```
+
+Set `maxRetries(0)` to disable internal retry entirely (every transient AWS error then surfaces
+immediately to the application).
+
+#### Connection-pool sharing
+
+The default `new SageMakerTransportFactory(config)` constructor backs every factory instance with
+a **process-wide shared** `SageMakerRuntimeHttp2AsyncClient`, keyed by the parts of
+`SageMakerConfig` that affect the underlying Netty HTTP/2 client (region, max concurrency,
+connect/acquire timeouts). Multiple factories built with the same config fingerprint reuse one
+Netty event loop group and one connection pool — so naive code that constructs a fresh factory
+per stream still gets a single, well-behaved client underneath.
+
+Without sharing, every factory instantiates its own Netty pool, and a burst of N factories
+triggers N simultaneous TLS handshakes from N distinct Netty clients against the same SageMaker
+endpoint. Under high concurrency (100+ streams) the SageMaker HTTP/2 frontline silently drops a
+large fraction of those streams before they ever reach the model container — verified
+end-to-end with CloudWatch logs from a 400-stream burst test against a 1× ml.g6.2xlarge endpoint:
+without sharing, ~65% of streams never appeared in the Deepgram container's listen log; with
+sharing, the burst behaves the same as the canonical Python load-test harness.
+
+Lifecycle:
+
+| Constructor | Client backing | `factory.shutdown()` |
+|---|---|---|
+| `SageMakerTransportFactory(config)` | shared (lazy-init, keyed by config fingerprint) | no-op — call `SageMakerTransportFactory.shutdownAllSharedClients()` once at app shutdown to release Netty resources |
+| `SageMakerTransportFactory(config, smClient)` | caller-provided (BYO, used for testing or custom credential providers) | no-op — caller owns the client lifecycle |
+
+```java
+// At app shutdown — releases all shared Netty pools the plugin lazily created.
+Runtime.getRuntime().addShutdownHook(new Thread(SageMakerTransportFactory::shutdownAllSharedClients));
+```
+
 ### Custom AWS Client
 
 For custom credential providers, proxy configuration, or testing:

examples/src/main/java/com/deepgram/examples/FluxSageMakerExample.java

@@ -8,6 +8,7 @@
 import com.deepgram.resources.listen.v2.websocket.V2WebSocketClient;
 import com.deepgram.sagemaker.SageMakerConfig;
 import com.deepgram.sagemaker.SageMakerTransportFactory;
+import com.deepgram.types.ListenV2Model;
 
 import java.io.RandomAccessFile;
 import java.nio.ByteBuffer;
@@ -85,10 +86,10 @@ public static void main(String[] args) throws Exception {
             done.countDown();
         });
 
-        // Connect — V2 uses model name as string via additionalProperty
+        // Connect using the typed Flux model constant from the SDK.
         CompletableFuture<Void> connectFuture = wsClient.connect(
                 V2ConnectOptions.builder()
-                        .model("flux-general-en")
+                        .model(ListenV2Model.FLUX_GENERAL_EN)
                         .build());
         connectFuture.get(30, TimeUnit.SECONDS);
         System.out.println("Connected. Streaming audio...\n");

examples/src/main/java/com/deepgram/examples/LiveMicFluxSageMakerExample.java

@@ -9,6 +9,7 @@
 import com.deepgram.sagemaker.SageMakerConfig;
 import com.deepgram.sagemaker.SageMakerTransportFactory;
 import com.deepgram.types.ListenV2Encoding;
+import com.deepgram.types.ListenV2Model;
 import com.deepgram.types.ListenV2SampleRate;
 
 import javax.sound.sampled.AudioFormat;
@@ -100,7 +101,7 @@ public static void main(String[] args) throws Exception {
 
         CompletableFuture<Void> connectFuture = wsClient.connect(
                 V2ConnectOptions.builder()
-                        .model("flux-general-en")
+                        .model(ListenV2Model.FLUX_GENERAL_EN)
                         .encoding(ListenV2Encoding.LINEAR16)
                         .sampleRate(ListenV2SampleRate.of(16000))
                         .build());

pom.xml

@@ -69,7 +69,7 @@
         <dependency>
             <groupId>com.deepgram</groupId>
             <artifactId>deepgram-java-sdk</artifactId>
-            <version>0.2.1</version>
+            <version>0.4.0</version>
         </dependency>
         <dependency>
             <groupId>software.amazon.awssdk</groupId>

sagemaker-transport/build.gradle

@@ -1,6 +1,6 @@
 dependencies {
     // Deepgram Java SDK — provides DeepgramTransport / DeepgramTransportFactory interfaces
-    api 'com.deepgram:deepgram-java-sdk:0.2.1'
+    api 'com.deepgram:deepgram-java-sdk:0.4.0'
 
     // AWS SDK v2 — SageMaker Runtime HTTP/2 bidirectional streaming
     api platform('software.amazon.awssdk:bom:2.42.0')