Kafka Connect Failed to Flush: Offset Commit Timeout Fix

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:

[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:

[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.

# 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:

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:

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 SourceRecords, 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.

# 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):

{
  "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:

{
  "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.