# Java SDKs, now async‑native: `CompletableFuture` + Reactive Streams
Source: https://speakeasy.com/blog/release-java-async

import { Callout, Table } from "@/lib/mdx/components";



Java's ecosystem has steadily shifted toward asynchronous, event‑driven programming. **Reactive architectures are increasingly adopted by major platforms**[^1], while `CompletableFuture` is ubiquitous in modern APIs. SDKs should match this **async‑first** reality—without forcing teams to abandon familiar synchronous code.

## What's new
* **Async native**: `CompletableFuture<T>` returned by standard methods.
* **Streaming built‑in**: Reactive Streams `Publisher<T>` for pagination, SSE, JSONL, and file I/O.
* **Dual SDK**: Synchronous by default; opt in with `.async()` per call‑site or service.
* **Blob**: A light, framework‑agnostic byte‑stream abstraction for efficient uploads/downloads.

## Why async

Traditional blocking SDKs dedicate **one thread per in‑flight request**. That results in idle threads, wasted memory, and hard‑to‑tune pools. Especially under high concurrency and variable network latency. Async I/O scales with a small, fixed number of threads without sacrificing composition[^3] or [backpressure](https://www.reactivemanifesto.org/glossary#Back-Pressure)[^2]. For key reactive concepts, see the [Reactive Manifesto glossary](https://www.reactivemanifesto.org/glossary).

<Callout title="Async I/O is already the standard everywhere else" type="info">
  Python has `asyncio`, JavaScript has async/await with `Promise`, Go has goroutines, and Swift has async/await. **Async I/O is the de facto model** across modern languages. Java's ecosystem is rapidly moving in this direction—think of this as **Python's `asyncio` but for Java**, with the added benefits of strong typing and mature reactive ecosystems.
</Callout>
### When async provides the most value

<Table
  className="px-4 my-8"
  containerSize="md"
  data={[
    {
      scenario: "Compositional orchestration",
      description: "Workflows with multiple dependent or parallel API calls (microservices orchestration, data enrichment)",
      benefit: "Futures/streams model sequencing, fan‑out, retries, and timeouts declaratively instead of blocking threads"
    },
    {
      scenario: "Latency tolerance",
      description: "Remote services with unpredictable or spiky response times",
      benefit: "Async frees threads during waits, so slow calls don't exhaust the pool or block unrelated work"
    },
    {
      scenario: "Concurrency scaling",
      description: "Thousands of requests in parallel (web servers, batch processors)",
      benefit: "Decouples request count from thread count — small event pools service massive concurrency without blowing up memory"
    },
    {
      scenario: "Event‑driven pipelines",
      description: "Real‑time filtering, transformation, and routing (SSE, JSONL streams, pagination)",
      benefit: "Lean on mature reactive ecosystems—Project Reactor, RxJava, Akka Streams—with proven operators for complex event processing"
    },
    {
      scenario: "Resource efficiency",
      description: "Cloud or edge environments with strict memory/cost budgets",
      benefit: "Far fewer thread stacks are required — often 80–90% lower memory under load, enabling smaller, cheaper deployments"
    }
  ]}
  columns={[
    { key: "scenario", header: "Scenario" },
    { key: "description", header: "Description" },
    { key: "benefit", header: "Async Benefit" }
  ]}
/>

<Callout title="SDK design insight for API producers" type="info">
  Your developers have diverse needs: some build high-concurrency microservices that benefit from async patterns, while others prefer straightforward synchronous flows. **Don't make them choose at the SDK level**—cater to both paradigms in one unified SDK so they can adopt async where it adds value without abandoning familiar synchronous code. The most successful SDKs optimize for **developer choice**, not architectural dogma.
</Callout>
Here's how this works in practice. We expose both sync and async interfaces from **one** SDK:

```java
// Build once — defaults to synchronous behavior
TradingSDK sdk = TradingSDK.builder()
    .serverURL("https://api.trading.com")
    .apiKey("your-api-key")
    .build();

// Synchronous usage (existing behavior)
Portfolio portfolio = sdk.getPortfolio("user-123");
List<Trade> trades = sdk.getTrades(portfolio.getId());

// Asynchronous usage via .async()
CompletableFuture<List<Trade>> asyncTrades = sdk
    .async()
    .getPortfolio("user-123")
    .thenCompose(p -> sdk.async().getTrades(p.getId()));
```

<Callout title="Backward compatibility" type="success">
  The builder yields a synchronous SDK by default, so **no breaking changes**. Opt into async via `.async()`; the mode applies consistently across sub‑SDKs.
</Callout>
## Implementation deep‑dive

Our HTTP stack uses Java 11's `HttpClient` async APIs and [NIO.2](https://docs.oracle.com/javase/tutorial/essential/io/fileio.html) primitives ([`ByteBuffer`](https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/nio/ByteBuffer.html), [`AsynchronousFileChannel`](https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/nio/channels/AsynchronousFileChannel.html), [`AsynchronousSocketChannel`](https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/nio/channels/AsynchronousSocketChannel.html)) for end‑to‑end non‑blocking I/O.

```java
// The underlying HTTP client call
client.sendAsync(
    request,
    HttpResponse.BodyHandlers.ofPublisher());
// returns a CompletableFuture
```

Our SDKs go to lengths to ensure there is as little as possible thread blocking work. For standard JSON responses, we collect and decode bytes asynchronously using `thenApply/thenCompose`. For streams, we hook into **`Flow.Publisher<List<ByteBuffer>>`**.

### Async iterables via Reactive Streams

For async iterables (pagination, streaming responses), we represent them as **[Reactive Streams](https://www.reactive-streams.org/)** [`Publisher<T>`](https://github.com/reactive-streams/reactive-streams-jvm/blob/v1.0.4/README.md#1-publisher-code). Traditional Java collections or custom iterators aren't feasible for async scenarios—they can't express **non‑blocking [backpressure](https://www.reactivemanifesto.org/glossary#Back-Pressure)** or handle variable consumption rates gracefully. Using `Publisher<T>` is an increasingly common idiom in modern Java applications, providing seamless interoperability with mature reactive ecosystems like [Project Reactor](https://projectreactor.io/), [RxJava](https://github.com/ReactiveX/RxJava), [Akka Streams](https://doc.akka.io/libraries/akka-core/current/stream/index.html), [Vert.x](https://vertx.io/docs/vertx-reactive-streams/java/), and [Quarkus Mutiny](https://quarkus.io/guides/mutiny-primer). We keep dependencies light by implementing JDK‑native operators (map, mapAsync, concat, wrap, flatten) through custom publishers, subscribers, and subscriptions.

### Protocol‑aware streaming
* **SSE & JSONL**: Custom publishers bridge raw byte streams to protocol parsers, yielding typed events on backpressure‑aware `Publisher<T>`
* **Pagination**: A generic, non‑blocking pagination publisher drives page fetches using pluggable `ProgressTrackerStrategy` to parse cursors and termination conditions
### Retries, timeouts, and hooks

* **Async retries** use `ScheduledExecutorService` for non‑blocking exponential backoff with jitter
* **Timeouts/cancellation[^5]** are surfaced with `CompletableFuture#orTimeout`, `completeExceptionally`, and cancellation propagation to HTTP requests
* **Hook transformations** leverage `CompletableFuture` for zero‑blocking customization points
## Efficient payloads with `Blob`

`Blob` is our core abstraction for working with streams of `ByteBuffer`:

```java
// Factories
Blob.from(Paths.get("large-dataset.json"));    // File path
Blob.from(inputStream);                         // InputStream
Blob.from("text content");                      // String
Blob.from(byteArray);                           // byte[]
Blob.from(flowPublisherOfByteBufferLists);
// Flow.Publisher<List<ByteBuffer>>

// Consumption
blob.asPublisher();       // org.reactivestreams.Publisher<ByteBuffer>
blob.toByteArray();       // CompletableFuture<byte[]>
blob.toFile(targetPath);  // CompletableFuture<Path>
blob.toInputStream();     // InputStream bridge
```

**Multipart uploads** concatenate publishers—each part (form field or file) is its own stream. **Downloads** expose `Blob` so you can stream to disk without buffering whole payloads.

<Callout title="Streaming uploads for synchronous SDKs" type="info">
  Enable `enableStreamingUploads: true` to get the same memory-efficient upload capabilities
  ```java
  // Sync SDK with streaming uploads
  syncSDK.uploadDataset(Blob.from(
      Paths.get("./datasets/large-file.jsonl")));
  ```
</Callout>

```java
// Stream a large model to disk
CompletableFuture<Path> model = asyncMLSDK.downloadModel("bert-large.bin")
    .thenCompose(res -> res.body().toFile(Paths.get("./models/bert-large.bin")));

// Stream binary telemetry (JSONL/SSE similar)
var bytes = asyncIoTSDK.streamSensorData("factory-floor-1")
    .thenApply(res -> res.body().asPublisher());

// Use your favourite reactive libraries to react to stream
Flux.from(Mono.fromFuture(bytes))
    .flatMap(Flux::from)
    .buffer(1024)
    .map(this::parseBinaryTelemetry)
    .filter(r -> r.getTemperature() > CRITICAL_THRESHOLD)
    .subscribe(this::triggerAlert);

// Upload a 10GB dataset without memory pressure
CompletableFuture<UploadResponse> up = asyncDataSDK.uploadDataset(
    Blob.from(Paths.get("./datasets/customer-data-10gb.jsonl"))
);
```

<Callout title="Reactive Streams vs JDK Flow" type="info">
  We expose **Reactive Streams** types in the public API for ergonomic use with the wider ecosystem. Java 11+ HTTP APIs expose `Flow.Publisher` under the hood; we convert via [`FlowAdapters`](https://www.reactive-streams.org/reactive-streams-jvm-1.0.4-javadoc/org/reactivestreams/FlowAdapters.html) and flatten `List<ByteBuffer>` as needed. If you require `Flow.Publisher`, adapters are available in both directions.
</Callout>
## Reactive streaming patterns

### Reactor

```java
// Paginate users
Publisher<User> users = asyncCRMSDK.listUsers().asPublisher();
Flux.from(users)
    .filter(User::isActive)
    .flatMap(u -> sendWelcomeEmail(u).thenReturn(u))
    .subscribe();
```

### RxJava

```java
// Real-time logs (SSE/JSONL abstracted)
Publisher<LogEntry> logStream = asyncLoggingSDK.streamLogs();
Flowable.fromPublisher(logStream)
    .filter(log -> "ERROR".equals(log.getLevel()))
    .window(30, TimeUnit.SECONDS)
    .flatMapSingle(window -> window.toList())
    .subscribe(alertingService::sendAlert);
```

### Akka Streams

```java
// Batched notifications
Publisher<Notification> notifications = asyncNotificationSDK.streamNotifications();
Source.fromPublisher(notifications)
    .filter(n -> "HIGH".equals(n.getPriority()))
    .mapAsync(10, this::enrichWithUserData)
    .grouped(50)
    .runForeach(batch -> pushNotificationService.sendBulk(batch), system);
```

## On Virtual Threads

[Virtual threads](https://docs.oracle.com/en/java/javase/21/core/virtual-threads.html) make **synchronous** code scale far better by reducing the cost of blocking. They're a great fit for simple request/response flows and can be used with the synchronous SDK today. Our **async SDK** still adds value where you need **backpressure, streaming, cancellation, and composition across multiple concurrent I/O operations**—patterns that `CompletableFuture`/`Publisher` express naturally.
## For API Producers: the unified paradigm advantage

* **Broader reach**: Works for Spring MVC (sync) *and* WebFlux/Quarkus (async)
* **Migration flexibility**: Adopt async gradually without replacing integrations
* **Lower maintenance**: One codebase; two clean interfaces
* **Consistent DX**: Same auth, errors, and config everywhere

## Getting started

Enable the following flag in your `.speakeasy/gen.yaml`:

```yaml
java:
  asyncMode: enabled
```

Then run `speakeasy run` to regenerate your SDK.

For existing SDKs, see our [migration guide](https://www.speakeasy.com/docs/customize/java/java-async-migration) for step-by-step instructions on enabling async support for your SDKs.

## Looking ahead

This release lays the groundwork for the full spectrum of modern Java patterns—from cloud‑native microservices to real‑time streaming systems. Ready to unlock the benefits of non‑blocking I/O? **[Regenerate your Java SDK](https://speakeasy.com/docs/sdks/create-client-sdks)** and try it today.

---

[^1]: [Netflix API with RxJava](https://netflixtechblog.com/reactive-programming-in-the-netflix-api-with-rxjava-7811c3a1496a) | [The Rise of Reactive Programming in Java](https://web.archive.org/web/20241213112122/https://www.icertglobal.com/the-rise-of-reactive-programming-in-java-blog/detail) | [Java Reactive Programming: An In-Depth Analysis](https://www.rethink.de/en/blog/java-reactive-programming-an-in-depth-analysis) | [Reactive Programming is Not a Trend](https://www.thisdot.co/blog/reactive-programming-is-not-a-trend-why-the-time-to-adopt-is-now) | [Java Reactive Programming](https://sam-solutions.com/blog/java-reactive-programming/)

[^2]: [Spring WebFlux Overview](https://docs.spring.io/spring-framework/reference/web/webflux/new-framework.html) — Why reactive/non‑blocking scales with a small, fixed number of threads.

[^3]: **Composition** — The ability to chain and combine async operations declaratively, like `future.thenCompose()` or stream operators, without manual thread coordination.

[^4]: **Streaming** — Processing data as a continuous flow of elements rather than loading everything into memory at once, enabling efficient handling of large datasets or real-time data.

[^5]: **Cancellation** — The ability to interrupt or stop async operations gracefully, propagating cancellation signals through composed operations to free resources and avoid unnecessary work.