[KAFKA-16394] ForeignKey LEFT join propagates null value on foreignKey change

• Type: Bug
• Status: Patch Available
• Priority: Major
• Resolution: Unresolved
• Affects Version/s: 3.7.0
• Component/s: streams
• Assignee: Ayoub Omari
• Reporter: Ayoub Omari
• Created: 20/Mar/24 18:00
• Updated: 27/Sep/24 00:40

Description

We have two topics : left-topic[String, LeftRecord] and right-topic[String, String]

where LeftRecord :

 case class LeftRecord(foreignKey: String, name: String)

we do a simple LEFT foreign key join on left-topic's foreignKey field. The resulting join value is the value in right-topic.

 

Scenario1: change foreignKey

Input the following

rightTopic.pipeInput("fk1", "1")
rightTopic.pipeInput("fk2", "2") 

leftTopic.pipeInput("pk1", LeftRecord("fk1", "pk1"))
leftTopic.pipeInput("pk1", LeftRecord("fk2", "pk1"))

 

Expected result

KeyValue(pk1, 1)
KeyValue(pk1, 2)

 

Actual result

KeyValue(pk1, 1)
KeyValue(pk1, null)
KeyValue(pk1, 2)

 

A null is propagated to the join result when the foreign key changes

 

Scenario 2: Delete PrimaryKey

Input

rightTopic.pipeInput("fk1", "1")
rightTopic.pipeInput("fk2", "2")

leftTopic.pipeInput("pk1", LeftRecord("fk1", "pk1"))
leftTopic.pipeInput("pk1", null) 

 

Expected result

KeyValue(pk1, 1)
KeyValue(pk1, null) 

 

Actual result

KeyValue(pk1, 1)
KeyValue(pk1, null)
KeyValue(pk1, null) 

An additional null is propagated to the join result.

 

This bug doesn't exist on versions 3.6.0 and below.

 

I believe the issue comes from the line https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionSendProcessorSupplier.java#L134

where we propagate the deletion in the two scenarios above

 

Attaching the topology I used.

Issue Links

• is blocked by: KAFKA-16343 Improve tests of streams foreignkeyjoin package
• is duplicated by: KAFKA-16644 FK join emits duplicate tombstone on left-side delete
• links to: GitHub Pull Request #15607