> ## Content Index > Fetch the complete content index at: https://www.ggorantala.dev/llms.txt > Use this file to discover other available public pages before exploring further. # Kafka Connect Failed to Flush: Offset Commit Timeout Fix - URL: https://www.ggorantala.dev/kafka-connect-failed-to-flush-offset-commit-timeout-fix/ - Published: 2026-02-27T14:21:00.000Z - Updated: 2026-08-29T08:20:00.000Z - Description: Kafka Connect source task logs 'Failed to flush, timed out while waiting for producer to flush outstanding N messages'. Here's the real cause and the fix. - Author: Gopi Gorantala - Tags: kafka-errors, ofset-flush-timeout, kafka-source-connector, debezium, event-streaming, Java, kafka-connect This one shows up in worker logs the moment a source connector (JDBC, Debezium, FileStream, a custom `SourceTask`) starts pulling faster than the cluster can absorb. The connector keeps running, data keeps flowing, and then every 60 seconds you get a wall of `WARN`/`ERROR` about a flush that timed out — and offsets that quietly stop advancing. Left alone it turns into a duplicate-reprocessing bomb on the next restart. This is a **source-side offset-commit** problem. It has nothing to do with converters, `errors.tolerance`, or the DLQ — those live on the sink/deserialization path. Do not go tuning `ErrorHandlingDeserializer` or dead-letter topics for this; they will not move the needle. ## 1\. The Error The exact lines from the Connect worker log: ```text [2026-07-13 09:14:22,455] ERROR WorkerSourceTask{id=jdbc-source-orders-0} Failed to flush, timed out while waiting for producer to flush outstanding 5183 messages (org.apache.kafka.connect.runtime.AbstractWorkerSourceTask:512) [2026-07-13 09:14:22,456] ERROR WorkerSourceTask{id=jdbc-source-orders-0} Failed to commit offsets (org.apache.kafka.connect.runtime.SourceTaskOffsetCommitter:117) ``` On busier tasks you will also see the companion warning as the outstanding queue keeps growing: ```text [2026-07-13 09:15:22,461] WARN WorkerSourceTask{id=jdbc-source-orders-0} Couldn't commit processed log positions with the source database due to a concurrent offset flush; ... ``` Key facts: - **Source class:** `AbstractWorkerSourceTask` (Apache Kafka **3.5+**, after the KIP-618 refactor). On 2.x / early 3.x the same message comes from `WorkerSourceTask`. - **Committer:** `SourceTaskOffsetCommitter`, which fires on a timer every `offset.flush.interval.ms`. - **Versions where this appears:** every Connect version from 0.10 to 4.0, self-managed or Confluent Platform, standalone or distributed. - **Typical setup:** a source connector under real throughput — Debezium on a busy table, JDBC bulk/incrementing mode, or a custom `SourceTask` that greedily returns large `poll()` batches. The task is **not killed**. This is not the `Task threw an uncaught and unrecoverable exception` failure. The task stays `RUNNING`; it just fails to *commit offsets*, over and over. ## 2\. How to Reproduce It Single-broker KRaft cluster plus a Connect worker, with the flush timeout cranked down so any real load trips it. ```yaml # docker-compose.yml services: kafka: image: apache/kafka:3.9.0 ports: - "29092:29092" environment: KAFKA_NODE_ID: 1 KAFKA_PROCESS_ROLES: broker,controller KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093 KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093,EXTERNAL://:29092 KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,EXTERNAL://localhost:29092 KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1 KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1 CLUSTER_ID: "5L6g3nShT-eMCtK--X86sw" connect: image: confluentinc/cp-kafka-connect:7.9.0 depends_on: [kafka] ports: - "8083:8083" environment: CONNECT_BOOTSTRAP_SERVERS: kafka:9092 CONNECT_GROUP_ID: connect-cluster CONNECT_CONFIG_STORAGE_TOPIC: _connect-configs CONNECT_OFFSET_STORAGE_TOPIC: _connect-offsets CONNECT_STATUS_STORAGE_TOPIC: _connect-status CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 1 CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: 1 CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: 1 CONNECT_KEY_CONVERTER: org.apache.kafka.connect.storage.StringConverter CONNECT_VALUE_CONVERTER: org.apache.kafka.connect.storage.StringConverter CONNECT_REST_ADVERTISED_HOST_NAME: connect CONNECT_PLUGIN_PATH: /usr/share/java,/usr/share/confluent-hub-components # The lever that makes this fire on demand: CONNECT_OFFSET_FLUSH_TIMEOUT_MS: 1 CONNECT_OFFSET_FLUSH_INTERVAL_MS: 5000 ``` Bring it up and register the built-in `FileStreamSource` pointed at a file you feed aggressively: ```bash docker compose up -d # Register a source that streams a file into a topic curl -s -X POST http://localhost:8083/connectors -H 'Content-Type: application/json' -d '{ "name": "file-source", "config": { "connector.class": "org.apache.kafka.connect.file.FileStreamSourceConnector", "tasks.max": "1", "file": "/tmp/feed.txt", "topic": "file-lines" } }' # Pump a large burst so the producer has thousands of in-flight records docker compose exec connect bash -c 'yes "line-of-data-$(date +%s%N)" | head -n 500000 > /tmp/feed.txt' ``` With `offset.flush.timeout.ms=1`, the committer cannot possibly drain the outstanding producer queue in 1 ms, so every interval logs: ```text ERROR WorkerSourceTask{id=file-source-0} Failed to flush, timed out while waiting for producer to flush outstanding 41120 messages ERROR WorkerSourceTask{id=file-source-0} Failed to commit offsets ``` In production you do not set `offset.flush.timeout.ms=1`. You leave it at the **default 5000 ms** and the *load* is what blows past it — a Debezium snapshot, a JDBC `bulk` pull of a big table, or a broker that briefly goes under-replicated and slows down acks. ## 3\. Why It Happens — Surface Level A source task's produce path and its offset-commit path are decoupled. The task calls `poll()`, gets a batch of `SourceRecord`s, and hands them to an internal Kafka producer that sends them asynchronously. Separately, on a timer (`offset.flush.interval.ms`, default **60000 ms**), the committer tries to record "we have durably produced everything up to source offset X" into the `_connect-offsets` topic. Committing an offset is only safe **after** every record before it has been acknowledged by the broker. So the commit first blocks, waiting for the producer to flush all outstanding (unacknowledged) sends — bounded by `offset.flush.timeout.ms`, default **5000 ms**. If the producer still has records in flight when that timeout expires, the flush is cancelled, the offsets are *not* committed, and they are restored to be retried on the next interval. Mechanically: **the source is producing records faster than the broker can acknowledge them, so the in-flight queue never drains inside the timeout window.** ## 4\. Why It Happens — Under the Hood `AbstractWorkerSourceTask` keeps a map of outstanding records — every `SourceRecord` sent to the producer that has not yet had its `Callback` fire. On each poll it converts records, calls `producer.send(record, callback)`, and adds them to that outstanding set; the callback removes them on ack. When `SourceTaskOffsetCommitter` triggers, `AbstractWorkerSourceTask.commitOffsets()` snapshots the current position, then **waits for the outstanding set to reach zero**, using `offset.flush.timeout.ms` as its deadline. The wait is effectively a `producer.flush()` scoped to this task's sends. If the deadline hits first, it logs `Failed to flush, timed out while waiting for producer to flush outstanding N messages`, aborts the commit, and leaves the source offset where it was. So the timeout is a symptom of **producer backpressure**. The producer has a finite `buffer.memory` (32 MB default) fronting a `RecordAccumulator`. When the drain rate (broker acks) can't keep up with the inflow rate (source `poll()` returning huge batches), records pile up as unacked in-flight requests. Every layer that slows the drain rate feeds this: - **`acks=all` with a slow ISR** — each send waits for `min.insync.replicas` to persist. One lagging follower stretches ack latency across the whole batch. - **Large records** — a source emitting multi-hundred-KB payloads fills `buffer.memory` in far fewer messages and pushes `max.request.size` / batch limits. - **A greedy connector** — JDBC in `bulk` mode or a snapshotting CDC connector returns tens of thousands of records per `poll()` with no upstream throttle, so the outstanding set explodes on the first interval. - **Transient broker trouble** — a `LEADER_NOT_AVAILABLE` blip, under-replicated partitions, or request-quota throttling stalls acks; the producer's in-flight queue backs up and the very next flush times out. The consequence is not data loss — a source connector re-reads from the **last committed** source offset. It's **duplicates**: on restart or rebalance, everything produced since the last successful commit is replayed. With `offset.flush.timeout.ms` chronically too low, offsets may *never* advance, so a restart re-emits an unbounded backlog. That's the real damage. If you run **exactly-once source support** (`exactly.once.source.support=enabled`, KIP-618, Kafka 3.3+), offset commits become transactional and the producer sends plus the offset write commit atomically — but the same `offset.flush.timeout.ms` still bounds how long the transaction waits to flush, so the identical timeout still fires under backpressure. The difference is you get no duplicates, just a failed transaction that retries. ## 5\. The Fix There are two distinct moves: raise the deadline, or reduce the backpressure. Do the second one in production. **Quick unblock (POC / genuinely slow-but-healthy sink): raise the timeout.** ```properties # connect-distributed.properties (worker-level) # BEFORE offset.flush.timeout.ms=5000 # AFTER offset.flush.timeout.ms=30000 ``` This buys the producer time to drain. It's the right call only when the flush is close-but-failing (drain takes 6–8 s against a healthy cluster). It does nothing if the source out-produces the broker indefinitely — you're just timing out later with a bigger backlog. **Real fix (production): relieve producer backpressure per connector.** Connect lets you override the source producer with `producer.override.*` in the *connector* config (requires `connector.client.config.override.policy=All` on the worker, the default since 3.0): ```json { "name": "jdbc-source-orders", "config": { "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector", "tasks.max": "1", "connection.url": "jdbc:postgresql://db:5432/app", "mode": "incrementing", "incrementing.column.name": "id", "topic.prefix": "pg-", "batch.max.rows": "500", "producer.override.linger.ms": "50", "producer.override.batch.size": "65536", "producer.override.compression.type": "lz4", "producer.override.buffer.memory": "67108864", "producer.override.max.request.size": "2097152" } } ``` What each lever does: - **`batch.max.rows` (JDBC) / `max.batch.size` (Debezium)** — throttle the *source* so `poll()` returns fewer records per call. This is the highest-leverage change: it caps the inflow rate at the origin instead of downstream. - **`compression.type=lz4`** and larger **`batch.size`** \+ a little **`linger.ms`** — fewer, denser produce requests raise the drain rate. - `**buffer.memory**` — size it to the peak in-flight volume so `send()` doesn't block, but understand more buffer alone just delays the wall; pair it with a higher flush timeout. - `**max.request.size**` — raise it for large payloads (must be ≤ broker `message.max.bytes`). **Fix the cluster if the cluster is the bottleneck.** If `acks=all` is stalling on a lagging follower, restore ISR health (`min.insync.replicas` arithmetic, `replica.lag.time.max.ms`) before touching connector config. A flush timeout that started exactly when a broker went under-replicated is a broker problem wearing a Connect costume. ## 6\. Best Practices & The Better Design The class of issue disappears when the source can't out-run the sink. Design for that: ```json { "name": "jdbc-source-orders", "config": { "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector", "tasks.max": "3", "mode": "incrementing", "incrementing.column.name": "id", "topic.prefix": "pg-", "batch.max.rows": "500", "poll.interval.ms": "1000", "producer.override.compression.type": "lz4", "producer.override.linger.ms": "50", "producer.override.batch.size": "131072", "producer.override.buffer.memory": "67108864", "producer.override.acks": "all" } } ``` The principles: - **Bound the source, don't just enlarge the buffer.** A bounded `poll()` (`batch.max.rows`, `max.batch.size`) plus `poll.interval.ms` creates natural backpressure so the outstanding set stays small and every flush completes well inside `offset.flush.timeout.ms`. - **Right-size `offset.flush.timeout.ms` as a function of drain time**, not a magic number. Measure how long a full-buffer flush takes at peak and set the timeout to \~2× that, with headroom. - **Spread load with `tasks.max`** matched to partitions / source shards so no single task carries the whole firehose. - **Turn on exactly-once source support** (`exactly.once.source.support=enabled`) where duplicates are unacceptable — it makes the produce + offset-commit atomic, so a timeout costs you a retried transaction, never replayed records. - **Keep the internal topics healthy.** `_connect-offsets` at RF≥3 / `min.insync.replicas=2` in production; a slow offsets topic causes the *commit* half to lag even when the produce half is fine. ## 7\. How to Prevent It Long-Term - **Monitor the source task metrics, not just the log.** From `kafka.connect:type=source-task-metrics`, watch `source-record-poll-rate` vs `source-record-write-rate` — a persistent gap is backpressure building. From `connector-task-metrics`, alert on `offset-commit-failure-percentage > 0` and rising `offset-commit-max-time-ms`. - **Watch the producer side.** The task's producer exposes `buffer-available-bytes` (heading toward zero = saturation) and `record-queue-time-avg` (climbing = the accumulator can't drain). These lead the timeout by minutes. - **Alert on the log signature.** Any sustained `Failed to flush, timed out while waiting for producer to flush outstanding` in worker logs is actionable — a burst during a broker restart self-heals, a continuous stream does not. - **Load-test the connector at real throughput** before production, including a CDC snapshot or full-table bulk pull, with `offset.flush.timeout.ms` at its production value. Assert offsets actually advance under load. - **Codify producer overrides per connector.** Ship `producer.override.*` and `batch.max.rows` as part of the connector definition in version control, reviewed like any other config, so nobody deploys a greedy unthrottled source by default. ## 8\. Key Takeaways - `Failed to flush, timed out while waiting for producer to flush outstanding N messages` is a **source-side offset-commit** problem — the task is `RUNNING`, offsets just stop advancing. It is unrelated to converters, `errors.tolerance`, or the DLQ. - The offset commit waits for the producer to drain all in-flight records within `offset.flush.timeout.ms` (**default 5000 ms**), fired every `offset.flush.interval.ms` (**default 60000 ms**). It's producer backpressure: the source out-produces the broker's ack rate. - The consequence is **duplicate reprocessing** on restart (source connectors re-read from the last committed offset), not data loss — and a chronically failing flush means offsets never advance, so the replay grows unbounded. - Raising `offset.flush.timeout.ms` is a POC bandaid. The real fix is reducing inflow (`batch.max.rows` / `max.batch.size`) and raising drain rate (`producer.override.*`: compression, batching, buffer). - Design the source to be bounded, size the timeout from measured drain time, use exactly-once source support where duplicates hurt, and alert on `offset-commit-failure-percentage` plus the producer's `buffer-available-bytes` — long before the log fills up.