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

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

     [ https://issues.apache.org/jira/browse/KAFKA-21276?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Eric Chang updated KAFKA-21276:
-------------------------------
    Description: 
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, a slow active task, the standby in its own JVM with {{-Xmx192m}})
* One input partition, 300,000 records of 1 KB: the standby fails with {{OutOfMemoryError: Java heap space}} after about 5 seconds, with about 139,000 records in the restore buffer.
* Eight input partitions, 12,000 records of 50 KB (about 600 MB): the standby fails with {{OutOfMemoryError: Java heap space}} after about 8 seconds, with about 2,250 records (about 113 MB) in the restore buffers of its standby partitions.

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.

  was:
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.


> 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
>            Priority: Major
>
> 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, a slow active task, the standby in its own JVM with {{-Xmx192m}})
> * One input partition, 300,000 records of 1 KB: the standby fails with {{OutOfMemoryError: Java heap space}} after about 5 seconds, with about 139,000 records in the restore buffer.
> * Eight input partitions, 12,000 records of 50 KB (about 600 MB): the standby fails with {{OutOfMemoryError: Java heap space}} after about 8 seconds, with about 2,250 records (about 113 MB) in the restore buffers of its standby partitions.
> 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)