Eric Chang created KAFKA-21276:
----------------------------------
Summary: Standby tasks with a source-topic changelog buffer
restore records without limit
Key: KAFKA-21276
URL: https://issues.apache.org/jira/browse/KAFKA-21276
Project: Kafka
Issue Type: Bug
Components: streams
Reporter: Eric Chang
Assignee: Eric Chang
A standby task whose state store uses the input topic as its changelog (for
example {{builder.table(topic)}} with {{topology.optimization=all}}) may only
apply records up to the offset committed by the active task.
{{StoreChangelogReader}} keeps fetching beyond that offset, and the records
that cannot be applied yet stay in {{ChangelogMetadata#bufferedRecords}}.
Nothing pauses the partition, so the buffer grows with every poll.
When the active task is far behind the end of the input topic, for example when
the application starts on a topic that already holds a lot of data, the standby
buffers almost the whole backlog and can run out of memory.
*Reproduction* (one broker, one input partition, 1 KB records, a slow active
task, standby in its own JVM with {{-Xmx192m}})
* 300,000 records in the input topic: the standby fails with
{{OutOfMemoryError: Java heap space}} after about 5 seconds, with about 139,000
records in the restore buffer.
* The same happens with 8 partitions and 50 KB records.
The same behavior can be shown without an OOM: with the active task paused (so
its committed offset does not move) and 10,000 new records in the input topic,
the standby fetched 8,703 of them within a few seconds, although it could not
apply any of them.
Other cases do not grow: a standby with its own changelog topic, and an active
task that is restoring (records that can be applied are applied right after
they are fetched).
*Proposed fix*
Pause such a changelog partition while its buffer holds records beyond the
limit offset, and resume it once the limit has moved past them. The buffer then
holds at most one fetch per partition ({{max.partition.fetch.bytes}}).
Related: KAFKA-19992 / KIP-1325, which proposes a configurable limit for the
restore buffer.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
