kumarpritam863 opened a new pull request, #15962: URL: https://github.com/apache/iceberg/pull/15962
## Summary
Adds a `BackpressureController` that prevents unbounded memory growth by
pausing Kafka Connect ingestion when buffered records exceed a configurable
high watermark, and resuming after commit flushes records to
disk.
Uses the standard `SinkTaskContext.pause()` / `SinkTaskContext.resume()`
API — the framework's built-in mechanism for sink connector backpressure.
### Motivation
The connector has zero backpressure. Records flow from `put()` through
`SinkWriter` into per-table `TaskWriter` instances with no bounds on the number
of concurrent writers. Each partition writer holds up to
128 MB in its Parquet row group buffer (verified by tracing through
`ParquetWriter.checkSize()` → `flushRowGroup()` at the row group size default).
With high-cardinality partitioning or dynamic routing to many
tables, the writer fan-out can exhaust heap:
Memory ≈ (active partition writers) × 128 MB (Parquet row group default)
Example: 10 tables × 50 partitions = 500 writers × 128 MB = ~64 GB
Additionally, during slow catalog operations (commit phase), new records
continue arriving unbounded since `put()` returns immediately.
### Design
Uses a **two-threshold watermark** pattern (same pattern as TCP flow
control, Netty channel backpressure, and Kafka's own `max.poll.records`):
put(batch) → SinkWriter.save(record) → write →
backpressure.recordBuffered()
↓
count >= highWatermark?
YES → context.pause(all
partitions)
Kafka Connect stops
polling consumer for those partitions
NO → continue
Worker.receive(START_COMMIT) → sinkWriter.completeWrite()
→ all writers flush to disk
→ backpressure.recordsFlushed()
↓
was paused?
YES → context.resume(all partitions)
Kafka Connect resumes polling on
those partitions
NO → continue
The hysteresis between high watermark (pause) and zero (resume after full
flush) prevents rapid pause/resume oscillation.
### New configuration
| Property | Type | Default | Description |
|----------|------|---------|-------------|
| `iceberg.backpressure.enabled` | Boolean | `false` | Enable backpressure
via pause/resume |
| `iceberg.backpressure.max-buffered-records` | Long | `1000000` | High
watermark: pause when this many records are buffered between commits |
**Off by default** — existing users who have tuned `commit.interval-ms`
and JVM heap are unaffected. Opt-in via:
```properties
iceberg.backpressure.enabled=true
iceberg.backpressure.max-buffered-records=500000
```
## Key design decisions
- Records, not bytes — TaskWriter (Iceberg internal, RollingFileWriter →
ParquetWriter) doesn't expose buffer size. Record count correlates with memory
and is cheap to track (AtomicLong.incrementAndGet() ≈
2ns/call).
- NOOP singleton — BackpressureController.NOOP has enabled=false and
highWatermark=Long.MAX_VALUE. Used in the no-arg SinkWriter constructor and all
tests. Counter still increments (for observability) but never
triggers pause.
- No Coordinator changes — backpressure is purely on the ingestion side.
The commit coordination protocol, control topic, and event format are
untouched.
- Thread safety — bufferedRecords is AtomicLong, paused is volatile.
pause()/resume() are idempotent (Kafka Connect handles duplicate calls
gracefully). No locking needed.
- Backward compatible constructor — SinkWriter(catalog, config) still
works, delegates to BackpressureController.NOOP. No existing code breaks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
