Upgrade from v1.0.6. This release bounds the Kafka consumer's read-ahead
and makes the peak a setting.
Bounded read-ahead:
- Kafka source: the depth of the channel between the poll goroutine and
the pipeline is now `fetch.prefetch`, and it defaults to 2 (#252). It
was a fixed 100, one whole fetch per slot, so a replay of a large
topic held the topic rather than a bounded slice of it. franz-go
bounds itself to one fetch per broker; sqlflow removed that bound.
Measured on a 3,000,000 message backlog, StructuredBatch at batch
5000, noop sink, medians of three runs: peak memory falls from
1,037 MiB to 285 MiB and throughput rises from 1,090,719 to
1,177,636 per second. At 10,000,000 messages, from 2,860 MiB to
293 MiB. Peak memory no longer tracks how long the pipeline was down.
Bounding costs no throughput. Every depth measured, 1 through 8, beat
the unbounded baseline, by 2.4% to 8.0%.
New configuration, all fields optional:
source:
type: kafka
kafka:
fetch:
max_bytes: 104857600 # per broker, unchanged default
max_partition_bytes: 10485760 # per partition, unchanged default
prefetch: 2 # fetches held ahead of the pipeline
The payload held is at most (prefetch + 2) x brokers x max_bytes, and
in practice (prefetch + 2) x partitions x max_partition_bytes. The
startup log prints all three, so the bound can be read out of the log.
The two byte limits count bytes on the wire, which are compressed
bytes. A topic that compresses well decompresses to more than the
limit in memory.
Behavior change to note when upgrading: a pipeline that relied on the
old 100 slot buffer to ride out broker stalls should set
`fetch.prefetch` higher. Raise it if a fast pipeline stalls waiting on
the broker; lower it to 1 when the container's memory limit is tight.
The `max_memory` and `max_state_size` limits in #162 are not in this
release. What changed is that the consumer's contribution to memory is
now a setting rather than a property of the outage.
