[
https://issues.apache.org/jira/browse/KAFKA-18713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17925431#comment-17925431
]
Matthias J. Sax commented on KAFKA-18713:
-----------------------------------------
{quote}Can I pick this up or is it bit too challenging for a newbie wanting to
contibute?
{quote}
You can always give it a try :) – It's hard to say how complicated a fix will
be, but yes, it's should deep inside the FK-join implementation...
{quote}If I don't create 10 partition, will the issue still occur? I tried to
reproduce this issue, with just one partition, everything works fine.
{quote}
Given the bug report, that is expected. I don't think that it must be 10
partitions, but it must be more than one partition to reproduce the bug.
Ideally, you would first write a new test case that reliably reproduces the
issue (for example extending `KTableKTableForeignKeyJoinDistributedTest` – not
sure if it's the best one to extend, but feel free to browse around in the code
base).
> Kafka Streams Left-Join not always emitting the last value
> ----------------------------------------------------------
>
> Key: KAFKA-18713
> URL: https://issues.apache.org/jira/browse/KAFKA-18713
> Project: Kafka
> Issue Type: Bug
> Components: streams
> Affects Versions: 3.8.0
> Reporter: tuna b
> Assignee: Nil Madhab
> Priority: Major
>
> There seems to be an issue when performing a left-join, the latest value with
> of the join does not contain the value of right side.
> {code:java}
> var builder = new StreamsBuilder();
> KTable<String, Department> departments = builder
> .table("departments",
> Materialized.<String,
> Department>as(Stores.persistentKeyValueStore("departments"))
> .withKeySerde(Serdes.String())
> .withValueSerde(CustomJsonSerde.json(Department.class)));
> KTable<String, Person> persons = builder
> .table("persons",
> Materialized.<String,
> Person>as(Stores.persistentKeyValueStore("persons"))
> .withKeySerde(Serdes.String())
> .withValueSerde(CustomJsonSerde.json(Person.class)));
> KTable<String, Person> joined = persons
> .leftJoin(departments, Person::getDepartmentId, (person, department) ->
> person.toBuilder()
> .department(department)
> .build(),
> TableJoined.as("my-joiner"),
> Materialized.<String,
> Person>as(Stores.persistentKeyValueStore("joined-results"))
> .withKeySerde(Serdes.String())
> .withValueSerde(CustomJsonSerde.json(Person.class)));
> joined
> .toStream()
> .to("joined-results", Produced.with(Serdes.String(),
> CustomJsonSerde.json(Person.class))); {code}
> How to reproduce:
> Create two topics persons and departments, each with 10 partitions.
> Pre-populate the departments topic with 2 departments.
>
> Observation: * When i initially produce a Person {{p-1}} with a FK
> {{{}dep-1{}}}, the join works .
> *
> ** output is an EnrichedResult with person {{p-1}} and department {{dep-1}}
> * When i change the FK to {{{}dep-2{}}}, the join updates .
> ** output is an EnrichedResult with person {{p-1 }}and department {{dep-2}}
> * When i change the FK back to {{{}dep-1{}}}, the join fails .
> ** output is an EnrichedResult with person {{p-1}} *but no department*
> * However, if I reproduce the same event ({{{}p-1{}}} with {{{}dep-1{}}}),
> the join works again .
> ** output is an EnrichedResult with person {{p-1}} and department {{dep-1}}
> Also, even when you are not setting back to a previous FK, there can still be
> an issue with the left join. Changing an FK means insert + delete operations,
> but sometimes the result of the delete is emitted after the result of the
> insert.
> How to reproduce:
> Create a departments topic and pre-populate it with 5 departments (dep-1 to
> dep-5). Create a persons topic and create person p-1 with FK dep-1. Send an
> update to the persons topic by changing the FK to dep-2 and repeat this step
> until dep-5. Now you will see that the latest emitted value of the person
> does not contain a department.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
