Hello, Kind reminder :)
Also, I want to add that I am trying to add a grace period for my kstream-ktable join, but I encounter an exception. It seems the Serde that I am providing in the join description is not taken into consideration when creating the *RocksDBTimeOrderedKeyValueBuffer*. *<I see some similar issue posted here: https://lists.apache.org/thread/0vnqkq61tg50rpc826fvq863b2nr96m2 <https://lists.apache.org/thread/0vnqkq61tg50rpc826fvq863b2nr96m2>>* *The code:* KStream<String, ContextAvro> contextKStream = streamsBuilder.stream(contextTopic); // Define the versioned state store supplier with a retention period VersionedBytesStoreSupplier storeSupplier = Stores.persistentVersionedKeyValueStore( contextStoreName, // Store name Duration.ofMinutes(2) // History retention period ); // Build the KTable with the versioned state store Materialized<String, ContextAvro, KeyValueStore<Bytes, byte[]>> materialized = Materialized.<String, ContextAvro>as( storeSupplier) .withKeySerde(Serdes.String()) .withValueSerde(getSpecificAvroSerde()); KTable<String, ContextAvro> contextKTable = contextKStream.toTable(materialized); FixedKeyProcessorSupplier<String, PaymentOperationAvro, ValueAndTrace<PaymentOperationAvro>> valueAndTraceProcessorSupplier = KafkaTracingValueAndTraceProcessor::new; // ValuAndTrace is a custom class which encapsulated the message V value and some tracingData // ValuAndTraceSerde is its Serde KStream<String, PaymentOperationProcessingResult> processingResultStream = adyenPaymentOperationKStream .processValues(valueAndTraceProcessorSupplier) .join(contextKTable, operationJoiner, *Joined.with(Serdes.String(), new ValueAndTraceSerde<>(getSpecificAvroSerde()), getSpecificAvroSerde(), null, Duration.ofSeconds(2)))* .peek(this::logFollowupOperation) .mapValues(operationHandler::handle); // operationJoiner is a custom joiner which decapsulate the value from ValueAndTrace back *The error:* org.apache.kafka.streams.errors.StreamsException: ClassCastException invoking processor: KSTREAM-JOIN-0000000039. Do the Processor's input types match the deserialized types? Check the Serde setup and change the default Serdes in StreamConfig or provide correct Serdes via method parameters. Make sure the Processor can accept the deserialized input of type key: java.lang.String, and value: com.nestle.nestpay.tracing.ValueAndTrace. Note that although incorrect Serdes are a common cause of error, the cast exception might have another cause (in user code, for example). For example, if a processor wires in a store, but casts the generics incorrectly, a class cast exception could be raised during processing, but the cause would not be wrong Serdes. at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:165) at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forwardInternal(ProcessorContextImpl.java:290) at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:269) at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:228) at org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:84) at org.apache.kafka.streams.processor.internals.StreamTask.lambda$doProcess$1(StreamTask.java:810) at org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.maybeMeasureLatency(StreamsMetricsImpl.java:872) at org.apache.kafka.streams.processor.internals.StreamTask.doProcess(StreamTask.java:810) at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:741) at org.apache.kafka.streams.TopologyTestDriver.completeAllProcessableWork(TopologyTestDriver.java:601) at org.apache.kafka.streams.TopologyTestDriver.pipeRecord(TopologyTestDriver.java:558) at org.apache.kafka.streams.TopologyTestDriver.pipeRecord(TopologyTestDriver.java:847) at org.apache.kafka.streams.TestInputTopic.pipeInput(TestInputTopic.java:115) at org.apache.kafka.streams.TestInputTopic.pipeInput(TestInputTopic.java:137) at com.nestle.nestpay.adyenadapter.KafkaStreamsServiceTest.whenAdyenFollowupOperationAndFollowupOperationContextEventsArePresentForTheSameKey_thenShouldConsumeTheEvents(KafkaStreamsServiceTest.java:279) at java.base/java.lang.reflect.Method.invoke(Method.java:568) at io.quarkus.test.junit.QuarkusTestExtension.runExtensionMethod(QuarkusTestExtension.java:1013) at io.quarkus.test.junit.QuarkusTestExtension.interceptTestMethod(QuarkusTestExtension.java:827) at java.base/java.util.ArrayList.forEach(ArrayList.java:1511) at java.base/java.util.ArrayList.forEach(ArrayList.java:1511) Caused by: java.lang.ClassCastException: class com.nestle.nestpay.tracing.ValueAndTrace cannot be cast to class org.apache.avro.specific.SpecificRecord (com.nestle.nestpay.tracing.ValueAndTrace and org.apache.avro.specific.SpecificRecord are in unnamed module of loader io.quarkus.bootstrap.classloading.QuarkusClassLoader @47c64cfe) at io.confluent.kafka.streams.serdes.avro.SpecificAvroSerializer.serialize(SpecificAvroSerializer.java:38) at org.apache.kafka.streams.state.internals.RocksDBTimeOrderedKeyValueBuffer.put(RocksDBTimeOrderedKeyValueBuffer.java:281) at org.apache.kafka.streams.kstream.internals.KStreamKTableJoinProcessor.process(KStreamKTableJoinProcessor.java:101) at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:152) ... 19 more" On Sun, 29 Dec 2024 at 09:45, Hagiu Alexandru <alexhagi...@gmail.com> wrote: > Hello, > > I'm developing a Quarkus microservice that utilizes Kafka Streams to > process messages from multiple topics. Specifically, I'm attempting to join > a KStream and a KTable derived from two of these topics. Under normal > conditions, the join operates as expected. However, under increased load, > the join occasionally fails to produce results, even though both messages > appear to be processed, but the ones from the KTable seems to be processed > slightly after the ones of the KStream(even if the actual timestamp of the > produced message is before). > > Code Example: > > KStream<String, OperationAvro> operationKStream = > streamsBuilder.stream(operationTopic); > KStream<String, ContextAvro> contextKStream = > streamsBuilder.stream(contextTopic); > KTable<String, ContextAvro> contextKTable = > contextKStream.toTable(Materialized.as(contextStoreName)); > > // both KTable and KStream use a custom processor which logs each message and > its timestamp > > KStream<String, Result> processingResultStream = operationKStream > .filter((key, value) -> isEligible(value)) > .join( > contextKTable, > (operation, context) -> topicsJoiner(operation, context), > Joined.with(Serdes.String(), new > ValueAndTraceSerde<>(getSpecificAvroSerde()), getSpecificAvroSerde()) > ) > .peek(this::logTheOutputOfJoiner) > .mapValues(handler::processTheJoinedData); > > > *Issue:* > > - Under normal load, the join between operationKStream and > contextKTable functions correctly. > - Under higher load, there are instances where the join does not > produce results, because when the KStream's message is processed, the > corresponding KTable's message is not yet present, even if it is actually > already sent from the producer. *Note:* when the KTable's message is > processed and logged(after the join was tried and skipped), the timestamp > of it(attached by the producer when the message was created) is before the > timestamp of the corresponding message from the KSTream. > > *Troubleshooting Steps Taken:* > > - Checked Partition Assignment: Ensured that messages with the same > key are assigned to the same partition across topics to facilitate correct > joining(co-partitioning). > - Adjusted num.stream.threads: Set to 6 as all the topics I'm > subscribed to have 6 partitions. This didn't solve the issue as the > messages with the same key will be distributed to the same stream task > which is on a single thread for co-partitioning. > - Adjusted max.task.idle.ms: Set to 2000 ms to allow for out-of-order > message handling( > > https://cwiki.apache.org/confluence/display/KAFKA/KIP-353%3A+Improve+Kafka+Streams+Timestamp+Synchronization). > This did not resolve the issue. > > My question is why this issue happen as the timestamp of the messages > seems to be in correct order, but it happen for the messages from the > KStream to be processed before the ones of KTable, which make the join to > be skipped. > > > Thank you, > > Alex Hagiu > >
