You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "Bruno Cadonna (Jira)" <ji...@apache.org> on 2020/06/19 15:47:00 UTC
[jira] [Resolved] (KAFKA-10148) Kafka Streams Restores too few
Records with eos-beta Enabled
[ https://issues.apache.org/jira/browse/KAFKA-10148?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Bruno Cadonna resolved KAFKA-10148.
-----------------------------------
Resolution: Fixed
> Kafka Streams Restores too few Records with eos-beta Enabled
> -------------------------------------------------------------
>
> Key: KAFKA-10148
> URL: https://issues.apache.org/jira/browse/KAFKA-10148
> Project: Kafka
> Issue Type: Bug
> Components: streams
> Reporter: Bruno Cadonna
> Assignee: Bruno Cadonna
> Priority: Blocker
> Fix For: 2.6.0
>
>
> System test {{StreamsEosTest.test_failure_and_recovery}} for eos-beta exposes a bug that results in wrong results in the output topic. The cause seems to be a too low end offset during restoration of a state store.
> Example:
> The system test computes a minimum aggregate over records in an input topic and writes the results to an output topic. The input topic partition {{data-1}} contains the following records among others:
> {code}
> ...
> offset: 125 CreateTime: 1591690264681 keysize: 5 valuesize: 4 sequence: 125 headerKeys: [] key: 14920 payload: 9215
> ...
> offset: 1611 CreateTime: 1591690297424 keysize: 5 valuesize: 4 sequence: 1611 headerKeys: [] key: 14920 payload: 1595
> ...
> offset: 2104 CreateTime: 1591690308542 keysize: 5 valuesize: 4 sequence: 2104 headerKeys: [] key: 14920 payload: 9274
> ...
> {code}
> The output topic partition {{min-1}} contains:
> {code}
> ...
> offset: 125 CreateTime: 1591690264681 keysize: 5 valuesize: 4 sequence: 125 headerKeys: [] key: 14920 payload: 9215
> ...
> offset: 1828 CreateTime: 1591690297424 keysize: 5 valuesize: 4 sequence: 1213 headerKeys: [] key: 14920 payload: 1595
> ...
> offset: 2324 CreateTime: 1591690308542 keysize: 5 valuesize: 4 sequence: 10 headerKeys: [] key: 14920 payload: 9215
> ...
> {code}
> The last record is obviously wrong because 1595 is less than 9215.
> To test the resilience to an unexpected failure of a Streams client, the system tests aborts a Streams client, i.e., the client is closed in a dirty manner. This dirty close causes the Streams client to restore its local state store that maintains the minimum aggregate from the beginning of the changelog topic partitions {{EosTest-KSTREAM-AGGREGATE-STATE-STORE-0000000003-changelog-1}}. The partition {{EosTest-KSTREAM-AGGREGATE-STATE-STORE-0000000003-changelog-1}} contains:
> {code}
> ...
> offset: 125 CreateTime: 1591690264681 keysize: 5 valuesize: 4 sequence: 125 headerKeys: [] key: 14920 payload: 9215
> ...
> offset: 1828 CreateTime: 1591690297424 keysize: 5 valuesize: 4 sequence: 1213 headerKeys: [] key: 14920 payload: 1595
> ...
> offset: 2324 CreateTime: 1591690308542 keysize: 5 valuesize: 4 sequence: 10 headerKeys: [] key: 14920 payload: 9215
> ...
> {code}
> Also here the last record is wrong.
> During the restoration, the Streams client uses its Kafka consumer to issue a list offsets request to get the end offset of the changelog topic partition. The response to the list offsets request contains end offset 1518 for {{EosTest-KSTREAM-AGGREGATE-STATE-STORE-0000000003-changelog-1}} as can be seen here:
> {code}
> [2020-06-09 08:11:49,250] DEBUG [Consumer clientId=EosTest-12216046-2d5d-48cd-864c-21b0aa570fae-StreamThread-1-restore-consumer, groupId=null] Received LIST_OFFSETS response from node 2 for request with header RequestHeader(apiKey=LIST_OFFSETS, apiVersion=5, clientId=EosTest-12216046-2d5d-48cd-864c-21b0aa570fae-StreamThread-1-restore-consumer, correlationId=3): (type=ListOffsetResponse, throttleTimeMs=0, responseData={EosTest-KSTREAM-AGGREGATE-STATE-STORE-0000000003-changelog-4=PartitionData(errorCode: 0, timestamp: -1, offset: 1478, leaderEpoch: Optional[0]), EosTest-KSTREAM-AGGREGATE-STATE-STORE-0000000003-changelog-1=PartitionData(errorCode: 0, timestamp: -1, offset: 1518, leaderEpoch: Optional[0])}) (org.apache.kafka.clients.NetworkClient)
> {code}
> Offset 1518 is before record in {{EosTest-KSTREAM-AGGREGATE-STATE-STORE-0000000003-changelog-1}}
> {code}
> offset: 1828 CreateTime: 1591690297424 keysize: 5 valuesize: 4 sequence: 1213 headerKeys: [] key: 14920 payload: 1595
> {code}
> Hence, this record is not restored into the local state store. However, after the restoration the input topic partition {{data-1}} is read starting with offset 2094. That means that record
> {code}
> offset: 1611 CreateTime: 1591690297424 keysize: 5 valuesize: 4 sequence: 1611 headerKeys: [] key: 14920 payload: 1595
> {code}
> is not read there either because it has a lower offset. Instead the following record with with key 14920 and value 9274 is read, but since 9274 is not less than 9215, value 9215 is written a second time to the output topic.
> I ran the system tests 10x with eos_alpha and 10x with eos_beta and only eos_beta failed a couple of times.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)