Hey there! Let's break down your questions one by one, drawing on Flink's design principles and real-world operational experience:

1. Should I always use async IO for database interactions? #

The short answer is no—it depends on your specific scenario:

• Use async IO when: Your database has high IO latency (e.g., remote/cloud databases, cross-region connections) or supports async drivers (like PostgreSQL's async JDBC). Async IO shines here because it avoids blocking Flink's operator threads while waiting for DB responses, letting those threads handle other tasks in the meantime.
• Stick with sync when: You're working with a low-latency local database (e.g., embedded H2, local PostgreSQL). The overhead of managing CompletableFutures, callbacks, and thread scheduling might outweigh the performance gains. In this case, sync code is simpler to write and maintain, and you can tune performance with appropriate parallelism.

Flink's AsyncFunction is designed specifically for high-latency IO scenarios—don't force it on low-latency use cases where sync code works just fine.

2. Is async (CompletableFuture) always more performant? How does sync + higher parallelism compare? #

Let's compare the two approaches:

• Async IO performance: In IO-bound scenarios, async IO delivers better throughput and resource efficiency with the same parallelism. Since operator threads aren't blocked waiting for IO, they can process more records per second. You also avoid the overhead of spawning excessive threads.
• Sync + higher parallelism: To match async IO's throughput, you'd need to crank up the parallelism significantly. Each sync operator instance will block its thread during IO calls, so more threads mean more concurrent IO requests. However, this comes with downsides:
  • Higher resource consumption (more threads, more DB connections—risking hitting your DB's connection pool limits).
  • More complex tuning (you have to balance parallelism with DB capacity to avoid overwhelming it).
  • Worse resource utilization (threads spend most of their time waiting instead of processing data).

For example, if you're calling a remote API with 500ms latency, async IO with parallelism 10 will outperform sync with parallelism 20—while using fewer resources. For a local DB with 1ms latency, sync with parallelism 10 might perform just as well as async, with simpler code.

3. Does the "Mailbox is in state CLOSED..." exception relate to too many threads? #

This exception is closely tied to thread lifecycle management, and excessive custom threads are a common culprit. Here's why:

• Flink uses a Mailbox mechanism to manage operator task scheduling and message passing. When an operator is shutting down (e.g., job is stopping, operator is restarting after failure), its mailbox transitions to the CLOSED state. Any attempt to submit tasks to a closed mailbox triggers this error.
• If you're using a custom thread pool with too many threads, those threads might still be running async callbacks (like DB response handlers) after the operator has started shutting down. When those callbacks try to send results back to the operator via the mailbox, they hit the closed state.

Fixes to try: #

• Use Flink-managed executors: Instead of creating your own thread pool, use getRuntimeContext().getExecutor()—Flink manages these threads and ensures they're shut down properly when the operator stops.
• Check operator state in callbacks: Before submitting tasks to the mailbox, verify if the operator is still running using isRunning() (available in AsyncFunction implementations).
• Limit async task concurrency: Use Flink's AsyncDataStream.unorderedWait() or orderedWait() with a reasonable capacity parameter to avoid flooding the system with too many pending async requests.

内容的提问来源于stack exchange，提问作者Xperia