[jira] [Created] (KAFKA-21276) Standby tasks with a source-topic changelog buffer restore records without limit

["Eric Chang (Jira)" <ji...@apache.org>]

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)