[KAFKA-12508] Emit-on-change tables may lose updates on error or restart in at_least_once

• Type: Bug
• Status: Resolved
• Priority: Blocker
• Resolution: Fixed
• Affects Version/s: 2.6.0, 2.7.0, 2.6.1
• Fix Version/s: 2.8.0, 2.7.1, 2.6.2
• Component/s: streams
• Assignee: John Roesler
• Reporter: Nico Habermann
• Created: 19/Mar/21 05:06
• Updated: 17/May/21 21:41
• Resolved: 25/Mar/21 23:10

Description

KIP-557 added emit-on-change semantics to KTables that suppress updates for duplicate values.

However, this may cause data loss in at_least_once topologies when records are retried from the last commit due to an error / restart / etc.

 

Consider the following example:

streams.table(source, materialized)
.toStream()
.map(mayThrow())
.to(output)

 

1. Record A gets read
2. Record A is stored in the table
3. The update for record A is forwarded through the topology
4. Map() throws (or alternatively, any restart while the forwarded update was still being processed and not yet produced to the output topic)
5. The stream is restarted and "retries" from the last commit
6. Record A gets read again
7. The table will discard the update for record A because
   1. The value is the same
   2. The timestamp is the same
8. Eventually the stream will commit
9. There is absolutely no output for Record A even though we're running in at_least_once

 

This behaviour does not seem intentional. The emit-on-change logic explicitly forwards records that have the same value and an older timestamp.

This logic should probably be changed to also forward updates that have an older or equal timestamp.

Issue Links

• is caused by: KAFKA-10248 Drop idempotent KTable source updates
• is caused by: KAFKA-8770 Either switch to or add an option for emit-on-change
• links to: GitHub Pull Request #10360
• links to: GitHub Pull Request #10397