[
https://issues.apache.org/jira/browse/KAFKA-20893?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Matthias J. Sax updated KAFKA-20893:
------------------------------------
Description:
KIP-1035 replaced the local state directory .checkpoint files with persistent
state store internal offset tracking, plus an in-memory offsets cache inside
StateDirectory. However, the cache is incorrectly populated with in-memory
store offsets, while it should only contain persistent store offsets (cf
[https://github.com/apache/kafka/commit/5740a26525a27df9eacd847a6d4ed6eb23fda0dc]
which replaces `ProcessorStateManger#checkpoint()` which contains a check
`storeMetadata.stateStore.persistent()` with new code lacking such a check).
After a task (with an in-memory) store is closed (and the in-memory state is
gone), these offset are still in the in-memory cache, and might get reported to
the task-assignor during a rebalance, which the task assignor would interpret
as "existing local state", even if there is none. This may leads to incorrect
task placement decisions.
The bug can only hit if a task has both an in-memory store and persistent
store, so by itself it should hit not too frequently, as most apps either have
all in-memory or all persistent stores (even if `suppress()` which is only
available in-memory elevate the problem).
However, with the recent changes via KIP-1071 (not release yet), this bug will
be elevated and thus it should be fixed before AK 4.4 gets released, by
ensuring that in-memory offsets are never added to the cache in the first place.
was:
KIP-1035 replace the local state directory .checkpoint files with persistent
state store internal offset tracking, plus an in-memory offsets cache inside
StateDirectory. However, the cache is incorrectly populated with in-memory
store offsets, while it should only contain persistent store offsets (cf
[https://github.com/apache/kafka/commit/5740a26525a27df9eacd847a6d4ed6eb23fda0dc]
which replaces `ProcessorStateManger#checkpoint()` which contains a check
`storeMetadata.stateStore.persistent()` with new code lacking such a check).
After a task (with an in-memory) store is closed (and the in-memory state is
gone), these offset are still in the in-memory cache, and might get reported to
the task-assignor during a rebalance, which the task assignor would interpret
as "existing local state", even if there is none. This may leads to incorrect
task placement decisions.
The bug can only hit if a task has both an in-memory store and persisted store,
so by itself it should hit not too frequently, as most app either have all
in-memory or all persistent (even if `suppress()` which is only available
in-memory elevate the problem).
However, with the recent changes via KIP-1071 (not release yet), this bug will
be elevated and thus it should be fixed before AK 4.4 gets released, by
ensuring that in-memory offsets are never added to the cache in the first place.
> KafkaStreams incorrectly reports task-offsets (ie state) for non-existing
> (previously owned) in-memory stores to the task assignor
> ----------------------------------------------------------------------------------------------------------------------------------
>
> Key: KAFKA-20893
> URL: https://issues.apache.org/jira/browse/KAFKA-20893
> Project: Kafka
> Issue Type: Bug
> Components: streams
> Affects Versions: 4.3.0
> Reporter: Matthias J. Sax
> Assignee: Matthias J. Sax
> Priority: Critical
> Fix For: 4.4.0, 4.3.2
>
>
> KIP-1035 replaced the local state directory .checkpoint files with persistent
> state store internal offset tracking, plus an in-memory offsets cache inside
> StateDirectory. However, the cache is incorrectly populated with in-memory
> store offsets, while it should only contain persistent store offsets (cf
> [https://github.com/apache/kafka/commit/5740a26525a27df9eacd847a6d4ed6eb23fda0dc]
> which replaces `ProcessorStateManger#checkpoint()` which contains a check
> `storeMetadata.stateStore.persistent()` with new code lacking such a check).
> After a task (with an in-memory) store is closed (and the in-memory state is
> gone), these offset are still in the in-memory cache, and might get reported
> to the task-assignor during a rebalance, which the task assignor would
> interpret as "existing local state", even if there is none. This may leads to
> incorrect task placement decisions.
> The bug can only hit if a task has both an in-memory store and persistent
> store, so by itself it should hit not too frequently, as most apps either
> have all in-memory or all persistent stores (even if `suppress()` which is
> only available in-memory elevate the problem).
> However, with the recent changes via KIP-1071 (not release yet), this bug
> will be elevated and thus it should be fixed before AK 4.4 gets released, by
> ensuring that in-memory offsets are never added to the cache in the first
> place.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
