[jira] [Comment Edited] (KAFKA-20792) Kafka Streams FK LeftJoin - IllegalArgumentException: Invalid partition: -1

["Siddhartha Devineni (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/KAFKA-20792?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18099860#comment-18099860 ] 

Siddhartha Devineni edited comment on KAFKA-20792 at 7/28/26 9:49 PM:
----------------------------------------------------------------------

[~pkleindl], 

Thank you for the reproducer repo; it made this straightforward to narrow down. The punctuator plus caching combination was the key detail;

The mechanism: StreamTask#punctuate sets a dummy ProcessorRecordContext with partition -1, the cache captures it at delete time and replays it at flush, and `SubscriptionSendProcessorSupplier` copies it into the subscription wrapper.


was (Author: JIRAUSER309328):
[~pkleindl], 

Thank you for the reproducer repo; it made this straightforward to narrow down. The punctuator plus caching combination was the key detail;

The mechanism: StreamTask#punctuate installs a dummy ProcessorRecordContext with partition -1, the cache captures it at delete time and replays it at flush, and `SubscriptionSendProcessorSupplier` copies it into the subscription wrapper.

> Kafka Streams FK LeftJoin - IllegalArgumentException: Invalid partition: -1
> ---------------------------------------------------------------------------
>
>                 Key: KAFKA-20792
>                 URL: https://issues.apache.org/jira/browse/KAFKA-20792
>             Project: Kafka
>          Issue Type: Bug
>          Components: streams
>    Affects Versions: 4.3.0
>            Reporter: Patrik Kleindl
>            Assignee: Siddhartha Devineni
>            Priority: Major
>
> Reproduced with https://github.com/pkleindl/kafka-streams-fk-exception/
> The topology has a foreign-key left join where the left-side state store has a Punctuator attached to perform regular cleanups of old records.
> After the punctuator deletes a record from the log, the stream crashes with "java.lang.IllegalArgumentException: Invalid partition: -1. Partition number should always be non-negative or null."
> The exception only happens if StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG is not 0.
> The resulting exception is:
> {code:java}
> 19:26:27.115 [fk-join-invalid-partition-it-31764876-b0bf-43a9-b0b1-7a75f89f3645-42a68a35-5e07-4b09-8353-a1828b9f5d3c-StreamThread-1] ERROR o.apache.kafka.streams.KafkaStreams - stream-client [fk-join-invalid-partition-it-31764876-b0bf-43a9-b0b1-7a75f89f3645-42a68a35-5e07-4b09-8353-a1828b9f5d3c] Encountered the following exception during processing and sent shutdown request for the entire application. org.apache.kafka.streams.errors.StreamsException: Exception caught in process. taskId=1_5, processor=join-subscription-join-foreign, topic=fk-join-invalid-partition-it-31764876-b0bf-43a9-b0b1-7a75f89f3645-join-subscription-registration-topic, partition=5, offset=0 at org.apache.kafka.streams.processor.internals.StreamTask.handleException(StreamTask.java:858) at org.apache.kafka.streams.processor.internals.StreamTask.handleException(StreamTask.java:840) at org.apache.kafka.streams.processor.internals.StreamTask.handleException(StreamTask.java:855) at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:822) at org.apache.kafka.streams.processor.internals.TaskExecutor.processTask(TaskExecutor.java:95) at org.apache.kafka.streams.processor.internals.TaskExecutor.process(TaskExecutor.java:76) at org.apache.kafka.streams.processor.internals.TaskManager.process(TaskManager.java:1934) at org.apache.kafka.streams.processor.internals.StreamThread.runOnceWithoutProcessingThreads(StreamThread.java:1238) at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:956) at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:916) Caused by: java.lang.IllegalArgumentException: Invalid partition: -1. Partition number should always be non-negative or null. at org.apache.kafka.clients.producer.ProducerRecord.<init>(ProducerRecord.java:77) at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.send(RecordCollectorImpl.java:259) at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.send(RecordCollectorImpl.java:175) at org.apache.kafka.streams.processor.internals.SinkNode.process(SinkNode.java:96) at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forwardInternal(ProcessorContextImpl.java:293) at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:272) at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:228) at org.apache.kafka.streams.kstream.internals.foreignkeyjoin.SubscriptionJoinProcessorSupplier$1.process(SubscriptionJoinProcessorSupplier.java:98) at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:179) at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forwardInternal(ProcessorContextImpl.java:293) at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:272) at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:228) at org.apache.kafka.streams.kstream.internals.foreignkeyjoin.SubscriptionReceiveProcessorSupplier$1.process(SubscriptionReceiveProcessorSupplier.java:96) at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:179) at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forwardInternal(ProcessorContextImpl.java:293) at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:272) at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:228) at org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:95) at org.apache.kafka.streams.processor.internals.StreamTask.lambda$doProcess$0(StreamTask.java:888) at org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.maybeMeasureLatency(StreamsMetricsImpl.java:955) at org.apache.kafka.streams.processor.internals.StreamTask.doProcess(StreamTask.java:888) at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:792) ... 6 common frames omitted{code}



--
This message was sent by Atlassian Jira
(v8.20.10#820010)