Definition
- AsyncFunction – A user-implemented function whose
asyncInvokemethod fires an external request and completes aResultFuturewhen the response returns, without blocking the operator thread. - AsyncDataStream.unorderedWait – Wires the AsyncFunction emitting results in whatever order responses complete, maximizing throughput while respecting watermark boundaries.
- AsyncDataStream.orderedWait – Preserves the original input order of records in the output by buffering early responses until predecessors finish.
- Capacity – The maximum number of in-flight requests, bounding memory and applying backpressure when the limit is reached.
- Timeout – The maximum time to wait for a response before triggering timeout handling.
- Checkpointing – In-flight requests are part of operator state so they are retried correctly on recovery, preserving exactly-once.
Imagine a clerk at the National Bank of India enriching each transaction by phoning an external credit bureau for a risk score. If the clerk dials, waits on hold for twenty seconds, hangs up, and only then picks up the next transaction, the queue crawls. Async I/O is like giving the clerk a bank of phone lines: they dial for the first transaction, and without waiting on hold they immediately dial for the next, so dozens of calls are ringing at once. As each bureau answers, the clerk records that score. The same twenty-second latency now overlaps across many transactions.
Understanding the Concepts
What is Flink's Async I/O?
Without Async I/O, a normal map or process function that calls an external system blocks the operator thread for the full round-trip. If each lookup takes twenty milliseconds, a single task can do at most fifty lookups per second. Async I/O decouples request issuance from response handling so many requests are outstanding at once.
package com.bank.enrichment;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import java.util.concurrent.TimeUnit;
public class AsyncIoDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<Transaction> transactions =
env.addSource(new KafkaSource<>("nbi-transactions"));
// Async I/O: enrich with credit bureau risk score
DataStream<EnrichedTransaction> enriched =
AsyncDataStream.unorderedWait(
transactions,
new CreditBureauAsyncFunction(),
30, TimeUnit.SECONDS, // timeout
100); // capacity: max 100 in-flight
enriched.addSink(new FraudScorerSink());
env.execute("NBI Async I/O Enrichment");
}
}
How do you implement an AsyncFunction?
You implement AsyncFunction whose asyncInvoke method receives an input record and a ResultFuture. Inside it you kick off the external call using a non-blocking client and complete the ResultFuture with the results when the response arrives.
package com.bank.enrichment;
import org.apache.flink.streaming.api.functions.async.AsyncFunction;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import java.math.BigDecimal;
import java.util.Collections;
import java.util.concurrent.CompletableFuture;
public class CreditBureauAsyncFunction
implements AsyncFunction<Transaction, EnrichedTransaction> {
private transient AsyncCreditBureauClient client;
@Override
public void open(Configuration parameters) {
// Initialize async HTTP client (non-blocking)
client = new AsyncCreditBureauClient("https://bureau.nbi.in");
}
@Override
public void asyncInvoke(Transaction txn,
ResultFuture<EnrichedTransaction> resultFuture) {
// Fire non-blocking request
CompletableFuture<Double> scoreFuture =
client.getRiskScore(txn.getAccountId());
// Complete ResultFuture when response arrives
scoreFuture.thenAccept(score -> {
EnrichedTransaction enriched = new EnrichedTransaction(
txn.getAccountId(),
txn.getAmount(),
txn.getEventTime(),
BigDecimal.valueOf(score));
resultFuture.complete(
Collections.singleton(enriched));
}).exceptionally(ex -> {
resultFuture.complete(
Collections.singleton(
txn.withDefaultScore())); // fallback
return null;
});
}
@Override
public void close() {
client.close();
}
}
What is the difference between unorderedWait and orderedWait?
| Feature | unorderedWait | orderedWait |
|---|---|---|
| Result order | Completion order (fastest first) | Original input order preserved |
| Throughput | Maximum (no head-of-line blocking) | Slightly lower (buffers early results) |
| Watermark handling | Respects watermark boundaries | Preserves strict order |
| Memory usage | Lower | Higher (buffers pending results) |
| Use when | Order does not matter downstream | Downstream depends on record order |
// unorderedWait: maximum throughput, results may arrive out of order
DataStream<EnrichedTransaction> unordered =
AsyncDataStream.unorderedWait(
transactions,
new CreditBureauAsyncFunction(),
30, TimeUnit.SECONDS, 100);
// orderedWait: preserves input order, buffers early responses
DataStream<EnrichedTransaction> ordered =
AsyncDataStream.orderedWait(
transactions,
new CreditBureauAsyncFunction(),
30, TimeUnit.SECONDS, 100);
How do capacity and timeout govern behavior?
Capacity caps the number of in-flight requests. When the limit is reached, the operator blocks new requests, creating natural backpressure. Timeout defines how long to wait for a response before calling the timeout handler.
package com.bank.enrichment;
import org.apache.flink.streaming.api.functions.async.AsyncFunction;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import java.util.Collections;
import java.util.concurrent.TimeoutException;
public class TimeoutHandlingDemo
implements AsyncFunction<Transaction, EnrichedTransaction> {
@Override
public void asyncInvoke(Transaction txn,
ResultFuture<EnrichedTransaction> resultFuture) {
client.getRiskScore(txn.getAccountId())
.thenAccept(score -> {
resultFuture.complete(Collections.singleton(
txn.withScore(score)));
});
}
@Override
public void timeout(Transaction txn,
ResultFuture<EnrichedTransaction> resultFuture) {
// Called when request exceeds timeout
// Option 1: complete with fallback value
resultFuture.complete(Collections.singleton(
txn.withDefaultScore())); // never lose the record
// Option 2: fail the job (strict mode)
// resultFuture.completeExceptionally(
// new TimeoutException("Bureau timeout for "
// + txn.getAccountId()));
}
}
Capacity and timeout behavior:
Capacity = 100, Timeout = 30s
Requests in flight: 98, 99, 100
--> new records WAIT (backpressure) until a slot opens
Request for ACC-001 sent at t=0
No response by t=30s
--> timeout() called
--> provide fallback score (no data loss)
What is the common mistake with Async I/O?
The actual asynchrony must come from a genuinely non-blocking client. Wrapping a blocking JDBC call in a thread pool inside asyncInvoke is a common mistake that does not give real concurrency and can exhaust threads.
// WRONG: wrapping blocking call in thread pool
@Override
public void asyncInvoke(Transaction txn,
ResultFuture<EnrichedTransaction> resultFuture) {
// This blocks a thread from the pool for each call
// With 100 capacity, you need 100 threads!
CompletableFuture.supplyAsync(() -> {
// BLOCKING JDBC call
return jdbcClient.querySync(txn.getAccountId());
}, threadPool).thenAccept(result -> {
resultFuture.complete(Collections.singleton(result));
});
}
// CORRECT: use a genuinely async client
@Override
public void asyncInvoke(Transaction txn,
ResultFuture<EnrichedTransaction> resultFuture) {
// Non-blocking: uses NIO under the hood
asyncHttpClient.getRiskScore(txn.getAccountId())
.thenAccept(score -> {
resultFuture.complete(Collections.singleton(
txn.withScore(score)));
});
}
How does Async I/O integrate with checkpointing?
In-flight requests are part of the operator's state. When a checkpoint occurs, the pending requests are stored in the snapshot. On recovery, Flink retries them correctly, preserving exactly-once semantics.
Checkpointing with Async I/O:
In-flight requests: [ACC-001, ACC-002, ACC-003]
Checkpoint triggered:
- Snapshot includes: pending requests + their input records
Checkpoint confirmed.
Crash and recovery:
- Restore from checkpoint
- In-flight requests [ACC-001, ACC-002, ACC-003] retried
- Each record processed exactly once in the output
What about the async lookup join in Flink SQL?
Flink SQL exposes the same capability declaratively through the async lookup join. You get Async I/O benefits without writing the AsyncFunction by hand.
-- Async lookup join in Flink SQL
SELECT t.account_id, t.amount, r.risk_score
FROM nbi_transactions AS t
JOIN credit_bureau FOR SYSTEM_TIME AS OF t.txn_time AS r
ON t.account_id = r.account_id;
-- The lookup table 'credit_bureau' is backed by an async connector
-- Flink automatically issues non-blocking lookups
-- No AsyncFunction code needed
// Equivalent SQL table definition:
// CREATE TABLE credit_bureau (
// account_id STRING,
// risk_score DOUBLE,
// PRIMARY KEY (account_id) NOT ENFORCED
// ) WITH (
// 'connector' = 'jdbc',
// 'lookup.async' = 'true',
// 'lookup.max-retries' = '3'
// );
What should a senior engineer know about Async I/O?
A senior engineer should set capacity to match the external system's concurrency limits, implement the timeout callback to provide a fallback rather than silently drop records, use unorderedWait as the default for throughput, and know that the async lookup join in Flink SQL provides the same capability declaratively.
package com.bank.enrichment;
public class SeniorAsyncIoChecklist {
// Senior engineer's Async I/O checklist:
//
// 1. Client MUST be genuinely non-blocking (NIO-based)
// 2. Capacity = external system's concurrency limit
// (too high overwhelms backend, too low underutilizes it)
// 3. Timeout > expected p99 latency of external system
// 4. Timeout handler: provide fallback, never silently drop
// 5. Default to unorderedWait for throughput
// 6. Use orderedWait only when downstream needs strict order
// 7. Monitor: in-flight count, timeout rate, latency p99
// 8. Consider Flink SQL async lookup join for simpler cases
}
Visual Overview
Synchronous vs Async I/O
Synchronous (blocking):
Record-1 ──▶ call bureau ──wait 20ms──▶ result ──▶ next
Record-2 ──▶ call bureau ──wait 20ms──▶ result ──▶ next
Record-3 ──▶ call bureau ──wait 20ms──▶ result ──▶ next
Total: 60ms for 3 records (sequential)
Async I/O (non-blocking):
Record-1 ──▶ call bureau ──┐
Record-2 ──▶ call bureau ──┤ all in flight simultaneously
Record-3 ──▶ call bureau ──┤
◀── response-2 ─┘ (completes first)
◀── response-1 ─┘
◀── response-3 ─┘
Total: ~20ms for 3 records (overlapped)
unorderedWait vs orderedWait:
Input order: R1, R2, R3
Completion: R2 (fast), R1 (medium), R3 (slow)
unorderedWait output: R2, R1, R3 (completion order)
orderedWait output: R1, R2, R3 (input order, R2 buffered)
Capacity and Backpressure:
Capacity = 3
┌────────────────────────────────────┐
│ In-flight: [R1] [R2] [R3] FULL │
│ R4 arrives: WAIT (backpressure) │
│ R2 completes: slot opens │
│ R4 can now be sent │
└────────────────────────────────────┘
Key Takeaway
Flink Async I/O is an operator pattern that lets you issue external requests asynchronously from within a stream operator so many requests are in flight concurrently instead of one blocking call at a time. You implement an AsyncFunction whose asyncInvoke method fires the request and completes a ResultFuture when the response arrives, and you wire it with AsyncDataStream.unorderedWait or orderedWait, specifying a timeout and a capacity.
The key benefit is that the operator thread does not block waiting for each response. Without Async I/O, a twenty-millisecond lookup caps a task at fifty lookups per second. With Async I/O, that latency overlaps across many records, so throughput rises dramatically. unorderedWait emits results in completion order for maximum throughput, while orderedWait preserves input order by buffering early results.
At the National Bank of India we enriched every transaction with a risk score from an external credit bureau whose API took about twenty milliseconds. Doing it synchronously throttled the enrichment stage badly. I rewrote it with Async I/O using the bureau's async client, so dozens of calls were in flight at once and the per-call latency overlapped, achieving roughly a tenfold throughput gain. For the fraud dashboard I used unorderedWait because order did not matter, but for the regulatory audit feed I used orderedWait.
The critical rules are that the client must be genuinely non-blocking, that capacity should match the external system's concurrency limits, and that the timeout callback must provide a fallback rather than silently drop records.
Remember one sentence: Async I/O overlaps external call latency across many records by issuing concurrent non-blocking requests via AsyncFunction with unorderedWait or orderedWait.
---