Re: Unexpected Tombstone records after kafka streams update 3.7.1

[Matthias J. Sax]

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