Backpressure

Slowness flows upstream

At the end of module 8, GatewayMeter reads gateway-events, runs parse, keys by request_id into assemble, keys the RequestRecords by tenant_id into tenant-cost, and writes TenantCostMinute through sink-tenant-cost with DeliveryGuarantee.EXACTLY_ONCE. This module inserts one operator between assemble and tenant-cost: enrich_price, which asks the billing service over HTTP for the tenant's negotiated price before the cost is computed. When that service is slow, the whole job runs at its speed. This page is about why that is by design, how you see it, and how you relieve it.

Records do not travel one at a time. Upstream's words: "Each record in Flink is sent to the next subtask compounded with other records in a network buffer, the smallest unit for communication between subtasks." Each subtask has "an input queue waiting to consume data and an output queue waiting to send data to the next subtask"; together that queued data is the job's in-flight data. A buffer is shipped when it is full or when execution.buffer-timeout expires — 100 ms by default (as of 2026-09; see upstream page). The pools are bounded: the target size of each is #channels * taskmanager.network.memory.buffers-per-channel + taskmanager.network.memory.floating-buffers-per-gate, with defaults of 2 exclusive buffers per channel and 8 floating buffers per gate, each buffer taskmanager.memory.segment-size bytes (32 KB in upstream's own example), as of 2026-09.

The bound is what makes backpressure. A receiver's free input buffers are what it can accept; the docs describe the exchange only in passing — "Assigning credits and sending buffers takes some time (around two roundtrip messages between two nodes)" — so read it as: the receiver announces how much it can take, the sender ships no more than that, and nothing is dropped or queued without limit. The details are in the network tuning guide at https://nightlies.apache.org/flink/flink-docs-release-2.3/docs/deployment/memory/network_mem_tuning/. The consequence is the definition upstream uses for the metric: "Internally, back pressure is judged based on the availability of output buffers. If a task has no available output buffers, then that task is considered back pressured."

Follow it through the job. enrich_price handles one record every two seconds, so its input buffers fill. Its upstream, assemble, can no longer send, so assemble's output buffers fill and it stops; its own input fills, parse stops, and the Kafka source, with no output buffer to write into, stops fetching. The connector's pendingRecords — "the number of records that have not been fetched by the source" — climbs, and so does consumer-group lag on the broker. Upstream: "Records in your job flow downstream (e.g. from sources to sinks) and back pressure is propagated in the opposite direction, up the stream."

What it does to checkpoints

Barriers ride the same buffers. "Barriers never overtake records, they flow strictly in line." Normally the checkpoint time is the snapshot; under backpressure, "the dominant factor in the end-to-end time of a checkpoint can be the time to propagate checkpoint barriers to all operators/subtasks." Two numbers in the checkpoint history show it, defined upstream: Start Delay, "the time it took for the first checkpoint barrier to reach this subtask since the checkpoint barrier has been created", and Alignment Duration, "the time between processing the first and the last checkpoint barrier. For aligned checkpoints, during the alignment, the channels that have already received checkpoint barrier are blocked from processing more data." The task-level metric checkpointStartDelayNanos carries the same signal: "A high value indicates back-pressure."

A checkpoint that takes longer than the timeout is "cancelled by the JobManager and a new checkpoint is triggered." The default value is on the upstream configuration page (https://nightlies.apache.org/flink/flink-docs-release-2.3/docs/deployment/config/); Module 7 already set it to 60 s with env.get_checkpoint_config().set_checkpoint_timeout(60000), which is why this experiment fits in fifteen minutes. And an expired checkpoint is not free. The tolerable checkpoint failure number "defines how many consecutive checkpoint failures will be tolerated, before the whole job is failed over. The default value is 0, which means no checkpoint failures will be tolerated, and the job will fail on first reported checkpoint failure." Expiration due to timeout is on the list of failures that count. So under module 7's configuration, a checkpoint timing out is a failover: restore checkpoint k, rewind to S_k, replay — into the same slow service.

Upstream gives three remedies and says they combine: remove the source of the backpressure, reduce the in-flight data, or take checkpoints unaligned. Buffer debloating is the second: set taskmanager.network.memory.buffer-debloat.enabled to true, and the mechanism "calculates the maximum possible throughput for the subtask (in the scenario that it is always busy) and adjusts the amount of in-flight data such that the consumption time of those in-flight data will be equal to the configured value", taskmanager.network.memory.buffer-debloat.target. Less data ahead of the barrier, faster alignment. It caps buffer size, not count, so it does not reduce memory.

Unaligned checkpoints are the third. They "contain in-flight data (i.e., data stored in buffers) as part of the checkpoint state, allowing checkpoint barriers to overtake these buffers. Thus, the checkpoint duration becomes independent of the current throughput." Enable them with env.get_checkpoint_config().enable_unaligned_checkpoints() or execution.checkpointing.unaligned: true; with execution.checkpointing.aligned-checkpoint-timeout set, "each checkpoint will still begin as an aligned checkpoint, but when the global checkpoint duration exceeds the aligned-checkpoint-timeout, if the aligned checkpoint has not completed, then the checkpoint will proceed as an unaligned checkpoint." Two cautions. Unaligned checkpointing "adds to I/O to the state storage, so you shouldn't use it when the I/O to the state storage is actually the bottleneck." And it does not cure the disease: unaligned checkpoints "can be used to speed up the propagation time of the checkpoint barriers. However please note, that this does not solve the underlying problem that's causing the backpressure in the first place (and end-to-end records latency will remain high)." The bill is still late; the checkpoints just stop failing.

Reading it

The Web UI's Back Pressure tab rates each subtask by the share of time it spent backpressured: "OK: 0% <= back pressured <= 10%", "LOW: 10% < back pressured <= 50%", "HIGH: 50% < back pressured <= 100%". Behind it are three metrics per subtask: backPressuredTimeMsPerSecond, idleTimeMsPerSecond, busyTimeMsPerSecond — "At any point of time these three metrics are adding up approximately to 1000ms." Idle excludes backpressured: "if the task is back pressured it is not idle."

The reading rule follows from the definitions. A backpressured task is a victim: it has work and cannot ship it. The cause is the most downstream task that is busy while everything upstream of it is backpressured — it has input, it has free output buffers, and it is not keeping up. In the job graph the source, parse and assemble turn black; enrich_price turns red; tenant-cost and the sink turn blue, idle, waiting for records that arrive at two per second. A synchronous HTTP call waiting on a socket is "busy" by this definition, which is why the metric is honest about it. The two flavours matter for checkpoints: softBackPressuredTimeMsPerSecond means the task is "still responsive and capable of for example triggering unaligned checkpoints", while under hard backpressure the task "is completely blocked and unresponsive preventing for example unaligned checkpoints from triggering."

Two more gauges confirm the buffers: inPoolUsage and outPoolUsage, "an estimate of the input buffers usage" and of the output buffers (which "can be > 100% if overdraft buffers are being used"). The external symptom is Kafka lag, and it has a twist. The source commits offsets to Kafka only on completed checkpoints, and "Committing offset is only for exposing the progress of consumer and consuming group for monitoring." With checkpoints expiring, the committed offset is frozen at S_57 even on partitions the source is still reading; broker-side lag overstates the backlog, and pendingRecords is the honest number. The consumer client's own metrics are forwarded under KafkaSourceReader.KafkaConsumer if you want them in the same dashboard.

Chains

"For distributed execution, Flink chains operator subtasks together into tasks. Each task is executed by one thread." Chaining "reduces the overhead of thread-to-thread handover and buffering": no serialization and no network buffers between parse and the source, or between chained maps. Flink chains by default where it can; some_stream.map(...).disable_chaining() keeps one operator out, some_stream.map(...).start_new_chain() begins a new chain at it, and slot_sharing_group("name") moves it to its own slot. When would you break one? Not for throughput — a two-second call in its own thread is still a two-second call. You break a chain when you need to see the operator: the three time metrics are per task, so a slow operator chained to assemble reports as assemble being busy. Upstream notes that the async operator "cannot be chained to a SourceFunction/SourceStreamTask" in any case.

Stage 9, first attempt: a synchronous map

class EnrichPriceSync(MapFunction):
    def open(self, ctx):
        self._http = BillingClient(BILLING_URL)          # blocking HTTP client

    def map(self, rec: RequestRecord) -> PricedRecord:
        price = self._http.get_price(rec.tenant_id)      # 20 ms today; 2 s during the incident
        return PricedRecord(rec, price)

priced = records.map(EnrichPriceSync(), output_type=PRICED_TYPE).uid("enrich-price")

Upstream describes exactly this shape: "Naively accessing data in the external database, for example in a MapFunction, typically means synchronous interaction: A request is sent to the database and the MapFunction waits until the response has been received. In many cases, this waiting makes up the vast majority of the function's time." The arithmetic is unforgiving. One record per call, one call at a time per subtask: at 2 s per call and parallelism 4, throughput is at most 4 / 2 s = 2 records per second, for the whole job. Scaling the map out helps a little at high cost: upstream warns that "many more parallel MapFunction instances means more tasks, threads, Flink-internal network connections, network connections to the database, buffers, and general internal bookkeeping overhead."

Stage 9, second attempt: async I/O

The idea, in upstream's words: "Asynchronous interaction with the database means that a single parallel function instance can handle many requests concurrently and receive the responses concurrently. That way, the waiting time can be overlaid with sending other requests and receiving responses." Three parts: an AsyncFunction that dispatches, a way to hand back the result ("await the result of the operation in Python API"), and the transformation applied with or without retry.

from typing import List

from pyflink.common import Time, Types
from pyflink.datastream import AsyncFunction, AsyncDataStream, async_retry_predicates
from pyflink.datastream.functions import RuntimeContext, AsyncRetryStrategy

class EnrichPrice(AsyncFunction[RequestRecord, PricedRecord]):
    def open(self, runtime_context: RuntimeContext):
        self._client = AsyncBillingClient(BILLING_URL)   # asyncio-native; must not block

    def close(self):
        self._client.close()

    async def async_invoke(self, rec: RequestRecord) -> List[PricedRecord]:
        price = await self._client.get_price(rec.tenant_id)
        return [PricedRecord(rec, price)]

    def timeout(self, rec: RequestRecord) -> List[PricedRecord]:
        # list price, flagged, rather than a dropped record: a re-rateable
        # accounting error beats a minute that is never billed
        return [PricedRecord(rec, PRICE_PER_1K[rec.model], price_source="list")]

retry = AsyncRetryStrategy.fixed_delay(
    max_attempts=3, backoff_time_millis=100,
    result_predicate=async_retry_predicates.empty_result_predicate,
    exception_predicate=async_retry_predicates.has_exception_predicate)

priced = AsyncDataStream.unordered_wait_with_retry(
    data_stream=records, async_function=EnrichPrice(),
    timeout=Time.seconds(5), async_retry_strategy=retry,
    capacity=100, output_type=PRICED_TYPE).uid("enrich-price")

The three parameters, as upstream defines them. Timeout "defines the maximum duration from the first invocation to the final completion of an asynchronous operation. This duration may include multiple retry attempts (if retries are enabled)"; by default "an exception is thrown and job is restarted", which is why timeout is overridden above — in Python "you can return a collection of results or raise an exception", and return [] drops the record. Capacity "defines how many asynchronous requests may be in progress per parallel instance (subtask) of the async operator at the same time"; it is the knob that bounds the queue, and "it will trigger backpressure once the capacity is exhausted." AsyncRetryStrategy "defines what conditions will trigger a delayed retry and the delay strategy". Retries need queue room: upstream's estimate is inputRate * retryRate * avgRetryDuration extra capacity.

The caveat that voids all of it: "the AsyncFunction is not called in a multi-threaded fashion. There exists only one instance of the AsyncFunction and it is called sequentially for each record", so "Using a database client whose lookup/query method call blocks until the result has been received back" or "Blocking/waiting on the future-type objects returned by an asynchronous client" gives you the synchronous map with extra steps. A synchronous client wrapped in a thread pool is the fallback, "usually less efficient than a proper asynchronous client."

Ordering. unordered_wait emits "as soon as the asynchronous request finishes"; ordered_wait "buffers a result record until all its preceding records are emitted (or timed out)", which "usually introduces some amount of extra latency and some overhead in checkpointing." Downstream of enrich_price is keyBy(tenant_id) into an event-time window, which places each record by its event timestamp, not by arrival — so unordered is safe here. It is safe with watermarks too, because in unordered mode "Watermarks do not overtake records and vice versa, meaning watermarks establish an order boundary"; the price of that is that "the unordered mode introduces some of the same latency and management overhead as the ordered mode does", proportional to watermark frequency. You would pay for ordered_wait only if a downstream operator relied on record order, and none in GatewayMeter does.

The cheaper design

The call is the problem, so make it rare. Module 3's PRICE_PER_1K is a constant; a negotiated price table changes a few times a day. Read the billing service's price changes as a broadcast stream into enrich_price (broadcast state, module 3), look the tenant up locally, and go to HTTP only on a miss. The async operator stays, for the misses, with a capacity sized for them. Note the spine's rule: the table is never fetched inside a keyed operator's process_element.

The failure, in numbers

Steady state: 120 RequestRecords per second, about 5 events each on gateway-events, checkpoints every 10 s, timeout 60 s, aligned. At t = 0 the billing service's p50 goes from 20 ms to 2 s.

Checkpoint 58's expiry fails the job over under the default tolerable failure number: state restores to checkpoint 57, offsets rewind to S_57, and the replay meets the same service; checkpoint 59 expires the same way, and the restart strategy from module 7 decides how long that loop runs. Raise the tolerable number and the job keeps running through expiring checkpoints — and the sink from module 8 makes the outcome the same: EXACTLY_ONCE output is committed on checkpoint completion, so a read_committed billing consumer sees nothing after checkpoint 57 either way. Event time at the sink advances at about one sixtieth of wall speed (2 of 120 records per second), so the 1-minute windows since t are open. This is the never-billed case, temporarily: the events are still in Kafka, S_57 is durable, and the minutes will be billed correctly — late. It becomes permanent only if the outage outlasts the topic's retention.

Now unordered_wait with capacity=100, same service, same 2 s:

The ceiling is capacity / latency. At a p50 of 5 s the same capacity gives 20 rec/s per subtask, 80 for the job, and backpressure returns — gently, and visible in the same metrics.

If a TaskManager dies right now

Upstream, on the async operator: "The asynchronous I/O operator offers full exactly-once fault tolerance guarantees. It stores the records for in-flight asynchronous requests in checkpoints and restores/re-triggers the requests when recovering from a failure." In this course's terms that is exactly-once state semantics, not end-to-end. So after recovery from checkpoint k, every request that was pending at k is issued again, and every record after S_k passes through enrich_price again. The billing lookup must be safe to repeat; it is a read, so it is. Had you put a write in that call — "record that this tenant was rated" — module 4 of the sibling course applies: the service sees it twice.

Survives: keyed state at k, including open requests and the window contents, and the in-flight records of enrich_price. Recomputed: everything from S_k forward, including the lookups. Emitted twice: nothing on tenant-cost, because the module-8 sink only commits on a completed checkpoint; alerts, still at-least-once, may repeat. Two footnotes for unaligned checkpoints: "Operators first recover the in-flight data before starting processing any data from upstream operators", and "on recovery, Flink generates watermarks after it restores in-flight data", so an operator that applies the latest watermark per record can behave differently than after an aligned checkpoint; GatewayMeter's windows do not.

Try it

Lab 9 is in labs/09-backpressure/:

• 9.1 — run stage 9 with a 2 s price lookup and an async capacity of 1 (GM_PRICE_DELAY_SECONDS=2 GM_ASYNC_CAPACITY=1); for two minutes record records in/out per second, busy and backpressured time, Kafka lag and checkpoint duration, and say where the backpressure starts and how far upstream it reaches.
• 9.2 — state the concurrency you need (throughput × service_time), then repeat with GM_ASYNC_CAPACITY=100 at the same input rate and delay; compare the same metrics, the timeout count and whether the output is complete.
• 9.3 — run the lab's tests: unordered completion may reorder records but must not change a request's identity or price; explain why the lookup would need an idempotency plan if it were a write.
• 9.4 — repeat once with GM_UNALIGNED=1; compare start delay, alignment time and checkpoint size, and record the in-flight data an unaligned checkpoint carries before calling it faster.

Lab page: lab-09-backpressure.html — the exercises, every source file, and the run-it-locally steps.