Yes, that's it. Thank you. Unfortunately, the issue is not fixed yet. There is 
a PR from March '24. This bug seems urgent to me. Are there any plans about 
merging the PR?

-Kevin

-----Ursprüngliche Nachricht-----
Von: Matthias J. Sax <mj...@apache.org> 
Gesendet: Freitag, 30. August 2024 04:19
An: users@kafka.apache.org
Betreff: Re: Unexpected Tombstone records after kafka streams update 3.7.1

Sounds like https://issues.apache.org/jira/browse/KAFKA-16394

-Matthias

On 8/29/24 02:35, Vogel, Kevin, DP EPS, BN, extern, external wrote:
> Hello there,
> 
> I searched the Apache Jira for a bug report on this topic but couldn't find 
> one. Maybe anyone else has noticed something similar or knows more about this.
> 
> After we updated our Spring Boot Kafka Streams application kafka-streams 
> dependency from 3.6.2 to 3.7.1, we noticed some failing tests. The expected 
> behavior of the streams-processor is, to join two input topics together and 
> produce a single output record. Then the test injects a third record with a 
> null value and expects another output record with a null value. But since the 
> update we are getting two records with null value instead of one.
> 
> I stripped down the processor to the absolute minimum, so you have some 
> example code. Don't worry about the strange names. We are working with a 
> legacy system and are unfortunately bound to them:
> 
> @Bean
> public BiFunction<
>      KStream<Gidpf01iKey, Gidpf01iValue>,
>      KTable<String, Gidpf01gAggregate>,
>      KStream<UuidString, Teil>>
> processTeil() {
>    return (gidpf01i, gidpf01gAggregate) -> {
>      var gidpf01iTable =
>          gidpf01i
>              .toTable();
> 
>      var joined =
>          gidpf01iTable.leftJoin(
>              gidpf01gAggregate,
>              new Gidpf01iGidpf01gAggregateForeignKeyExtractor(),
>              new Gidpf01iGidpf01gAggregateValueJoiner(),
>              TableJoined.as("gidpf01i-gidpf01g-aggregate-to-teil"));
> 
>      return joined.toStream().selectKey(new GenericSbamUuidGenerator<>());
>    };
> }
> 
> You see nothing fancy. Just a KTable - Ktable left-join. Even more strange 
> is, that not all streams-processors behave differently after the update. But 
> they are all very similar and I can not see any significant difference.
> 
> The test looks like this:
> 
>    @Test
>    void testProcessGidpf01iTombstone() {
> 
>      // Key und Value für Teilestamm und Teiletext
>      final var gidpf01i = createGidpf01iKeyValue("A", REFRESH);
>      final var gidpf01gAggregate =
>          createGidpf01gAggregate("T123456789", "Test Beschreibung", 
> "Test Zusatzinformation");
> 
>      gipf01gAggregateInputTopic.pipeInput(gidpf01gAggregate);
>      gidpf01iInputTopic.pipeInput(gidpf01i);
> 
>      // Datensatz gelesen
>      var results = teilOutputTopic.readKeyValuesToList();
>      assertThat(results).hasSize(1);
>      assertThat(results.get(0).key).isInstanceOf(UuidString.class);
>      assertThat(results.get(0).value).isNotNull();
> 
>      // Tombstone für GIDPF01I senden
>      gidpf01iInputTopic.pipeInput(new TestRecord<>(gidpf01i.key(), 
> null));
> 
>      results = teilOutputTopic.readKeyValuesToList();
>      assertThat(results).hasSize(1);                 // <--- failing here 
> because results contain two identical records with null value instead of one
>      assertThat(results.get(0).key).isInstanceOf(UuidString.class);
>      assertThat(results.get(0).value).isNull();
> 
>      assertThat(gidpf01iDlqOutputTopic.isEmpty()).isTrue();
>    }
> 
> Does anyone know about a bug or behavioral change like this in Kafka streams 
> 3.7.1? I'm very grateful for any response.
> 
> Kind Regards
> Kevin Vogel
> Software Developer extern
> Mitarbeiter der Qvest Digital AG
> 
> Am Dickobskreuz 10, D-53121 Bonn
> Tel.: +49 228 54881-0
> HRB AG Bonn 18196 Ust-ID (VAT): DE274355441
> Vorstand: Dr. Stefan Barth, Kai Ebenrett, Boris Esser, Alexander Steeg 
> Vorsitzender Aufsichtsrat: Stefan Nöthen
> 
> 

Reply via email to