KAFKA-20792: Fix invalid partition -1 in FK join triggered by punctuator

streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionSendProcessorSupplier.java

@@ -241,11 +241,24 @@ private void forward(final Record<KLeft, Change<VLeft>> record, final KRight for
                 hash(record),
                 deleteKeyNoPropagate,
                 record.key(),
-                context().recordMetadata().get().partition()
+                this.currentPrimaryPartition()
             );
             context().forward(record.withKey(foreignKey).withValue(wrapper));
         }
 
+        /**
+         * Returns the partition of the record currently being processed or null if there is no source record (for example,
+         * when the downstream forward was triggered by a punctuator, possibly indirectly via a state store cache flush).
+         * <p>
+         * A null primary partition instructs the subscription response sink to fall back to default partitioning.
+         */
+        private Integer currentPrimaryPartition() {
+            return context().recordMetadata()
+                    .map(RecordMetadata::partition)
+                    .filter(p -> p >= 0)
+                    .orElse(null);
+        }
+
         private long[] hash(final Record<KLeft, Change<VLeft>> record) {
             if (recordHash == null) {
                 recordHash = record.value().newValue == null

streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionWrapperSerde.java

@@ -34,6 +34,10 @@
 
 public class SubscriptionWrapperSerde<KLeft> extends WrappingNullableSerde<SubscriptionWrapper<KLeft>, KLeft, Void> {
 
+    // Sentinel for a null primaryPartition, which the fixed-width V1 field cannot otherwise express.
+    // Partitions are never negative, so -1 can never collide with a real value.
+    private static final int NULL_PRIMARY_PARTITION = -1;
+
     public SubscriptionWrapperSerde(final Supplier<String> primaryKeySerializationPseudoTopicSupplier,
                                     final Serde<KLeft> primaryKeySerde) {
         super(
@@ -158,7 +162,7 @@ private byte[] serializeV0(final SubscriptionWrapper<KLeft> data, final Headers
 
         private byte[] serializeV1(final SubscriptionWrapper<KLeft> data, final Headers headers) {
             final ByteBuffer buf = serializeCommon(data, headers, data.version(), Integer.BYTES);
-            buf.putInt(data.primaryPartition());
+            buf.putInt(data.primaryPartition() == null ? NULL_PRIMARY_PARTITION : data.primaryPartition());
             return buf.array();
         }
     }
@@ -224,7 +228,8 @@ public SubscriptionWrapper<KLeft> deserialize(final String ignored, final Header
             );
             final Integer primaryPartition;
             if (version > 0) {
-                primaryPartition = buf.getInt();
+                final int rawPrimaryPartition = buf.getInt();
+                primaryPartition = rawPrimaryPartition == NULL_PRIMARY_PARTITION ? null : rawPrimaryPartition;
             } else {
                 primaryPartition = null;
             }

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableForeignKeyJoinScenarioTest.java

@@ -24,6 +24,7 @@
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.apache.kafka.common.serialization.StringSerializer;
 import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.streams.KeyValue;
 import org.apache.kafka.streams.StreamsBuilder;
 import org.apache.kafka.streams.StreamsConfig;
 import org.apache.kafka.streams.TestInputTopic;
@@ -36,8 +37,15 @@
 import org.apache.kafka.streams.kstream.Materialized;
 import org.apache.kafka.streams.kstream.Produced;
 import org.apache.kafka.streams.kstream.TableJoined;
+import org.apache.kafka.streams.processor.PunctuationType;
 import org.apache.kafka.streams.processor.StreamPartitioner;
+import org.apache.kafka.streams.processor.api.FixedKeyProcessor;
+import org.apache.kafka.streams.processor.api.FixedKeyProcessorContext;
+import org.apache.kafka.streams.processor.api.FixedKeyRecord;
+import org.apache.kafka.streams.state.KeyValueIterator;
 import org.apache.kafka.streams.state.KeyValueStore;
+import org.apache.kafka.streams.state.TimestampedKeyValueStore;
+import org.apache.kafka.streams.state.ValueAndTimestamp;
 import org.apache.kafka.streams.test.TestRecord;
 import org.apache.kafka.streams.utils.UniqueTopicSerdeScope;
 import org.apache.kafka.test.StreamsTestUtils;
@@ -47,8 +55,11 @@
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
 
+import java.time.Duration;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Properties;
@@ -69,6 +80,7 @@ public class KTableKTableForeignKeyJoinScenarioTest {
     private static final String LEFT_TABLE = "left_table";
     private static final String RIGHT_TABLE = "right_table";
     private static final String OUTPUT = "output-topic";
+    private static final String LEFT_STORE = "left-store";
 
     @ParameterizedTest
     @ValueSource(booleans = {false, true})
@@ -485,4 +497,98 @@ public Optional<Set<Integer>> partitions(final String topic, final Integer key,
         assertTrue(leftPartitionerCalled.get());
         assertTrue(rightPartitionerCalled.get());
     }
+
+    /**
+     * Regression test for KAKFA-20792.
+     * <p>
+     * When a punctuator deletes a record from the materialized store backing the left-hand side of a foreign-key join,
+     * the {@code  StreamTask#punctuate} sets a dummy {@code ProcessorRecordContext} (null topic, -1 partition, -1 offset).
+     * The caching layer captures that dummy context and replays it when the change is flushed downstream. The subscription
+     * send processor injects that -1 into {@code SubscriptionWrapper#primaryPartition}, which later failed with
+     * {@code IllegalArgumentException: Invalid partition: -1} when the subscription response was routed back to the primary-key partition.
+     */
+    @Test
+    public void shouldNotFailWhenPunctuatorDeletesFromLeftTableStore() {
+        final String leftTopic = "left-topic";
+        final String rightTopic = "right-topic";
+        final String primaryKey = "pk";
+        final String foreignKey = "fk";
+
+        final StreamsBuilder builder = new StreamsBuilder();
+
+        final KTable<String, String> left = builder
+                .stream(leftTopic, Consumed.with(Serdes.String(), Serdes.String()))
+                .toTable(Materialized.as(LEFT_STORE));
+
+        final KTable<String, String> right = builder
+                .stream(rightTopic, Consumed.with(Serdes.String(), Serdes.String()))
+                .toTable(Materialized.as("right-store"));
+
+        left.leftJoin(
+                right,
+                value -> foreignKey,
+                (leftValue, rightValue) -> leftValue + "+" + rightValue,
+                Materialized.as("join-store"))
+                .toStream()
+                .to(OUTPUT, Produced.with(Serdes.String(), Serdes.String()));
+
+        // attach a wall-clock punctuator to the left table's own mateialized store
+        left.toStream().processValues(StoreCleaner::new, LEFT_STORE);
+
+        final Properties config = new Properties();
+        config.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, "fk-join-punctuation-test");
+        config.setProperty(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:1234");
+        config.setProperty(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getAbsolutePath());
+        config.setProperty(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
+        config.setProperty(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
+        // caching must be enabled: the cache flush listener is what replays the record context that was captured
+        // while the punctuator mutated the store
+        config.setProperty(StreamsConfig.STATESTORE_CACHE_MAX_BYTES_CONFIG, String.valueOf(10 * 1024 * 1024));
+
+        try (final TopologyTestDriver driver = new TopologyTestDriverBuilder(builder.build()).withConfig(config).build()) {
+            final TestInputTopic<String, String> leftInput = driver.createInputTopic(leftTopic, new StringSerializer(), new StringSerializer());
+            final TestInputTopic<String, String> rightInput = driver.createInputTopic(rightTopic, new StringSerializer(), new StringSerializer());
+            final TestOutputTopic<String, String> output = driver.createOutputTopic(OUTPUT, new StringDeserializer(), new StringDeserializer());
+
+            rightInput.pipeInput(foreignKey, "rightValue");
+            leftInput.pipeInput(primaryKey, "leftValue");
+
+            assertThat(output.readKeyValue(), is(KeyValue.pair(primaryKey, "leftValue+rightValue")));
+
+            // fires the punctuator, which deletes the left record and the subsequent cache flush forwards the change
+            // downstream while the dummy punctuation record context is in effect
+            driver.advanceWallClockTime(Duration.ofSeconds(2));
+
+            // the left join must emit a tombstone rather than failing or dropping the delete silently
+            assertThat(output.readKeyValue(), is(KeyValue.pair(primaryKey, null)));
+            assertTrue(output.isEmpty());
+        }
+    }
+
+    /**
+     * Deletes every record from {@link #LEFT_STORE} on each wall-clock-time punctuation.
+     */
+    private static final class StoreCleaner implements FixedKeyProcessor<String, String, String> {
+
+        private TimestampedKeyValueStore<String, String> store;
+
+        @Override
+        public void init(final FixedKeyProcessorContext<String, String> context) {
+            store = context.getStateStore(LEFT_STORE);
+            context.schedule(Duration.ofSeconds(1), PunctuationType.WALL_CLOCK_TIME, timestamp -> {
+                final List<String> keys = new ArrayList<>();
+                try (final KeyValueIterator<String, ValueAndTimestamp<String>> iterator = store.all()) {
+                    while (iterator.hasNext()) {
+                        keys.add(iterator.next().key);
+                    }
+                }
+                keys.forEach(store::delete);
+            });
+        }
+
+        @Override
+        public void process(final FixedKeyRecord<String, String> record) {
+            // no-op: this processor exists only to own the punctuator and connect the store
+        }
+    }
 }

streams/src/test/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionSendProcessorSupplierTest.java

@@ -38,6 +38,7 @@
 import static org.hamcrest.Matchers.empty;
 import static org.hamcrest.Matchers.greaterThan;
 import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.nullValue;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotEquals;
 
@@ -67,6 +68,38 @@ public class SubscriptionSendProcessorSupplierTest {
     private final String fk1 = "fk1";
     private final String fk2 = "fk2";
 
+    @Test
+    public void shouldUseNullPrimaryPartitionWhenRecordMetadataIsAPunctuationDummy() {
+        final MockInternalProcessorContext<String, SubscriptionWrapper<String>> context = new MockInternalProcessorContext<>();
+        leftJoinProcessor.init(context);
+        // A punctuator sets a dummy record context: null topic, -1 partition, -1 offset.
+        context.setRecordMetadata(null, -1, -1);
+
+        final LeftValue leftRecordValue = new LeftValue(fk1);
+
+        leftJoinProcessor.process(new Record<>(pk, new Change<>(leftRecordValue, null), 0));
+
+        assertThat(context.forwarded().size(), is(1));
+        assertThat(
+            context.forwarded().get(0).record(),
+            is(new Record<>(fk1, new SubscriptionWrapper<>(hash(leftRecordValue), PROPAGATE_NULL_IF_NO_FK_VAL_AVAILABLE, pk, null), 0))
+        );
+    }
+
+    @Test
+    public void shouldUseNullPrimaryPartitionWhenRecordMetadataIsAbsent() {
+        final MockInternalProcessorContext<String, SubscriptionWrapper<String>> context = new MockInternalProcessorContext<>();
+        leftJoinProcessor.init(context);
+        // no setRecorMetadata() call, so recordMetadata() is empty
+
+        final LeftValue leftRecordValue = new LeftValue(fk1);
+
+        leftJoinProcessor.process(new Record<>(pk, new Change<>(leftRecordValue, null), 0));
+
+        assertThat(context.forwarded().size(), is(1));
+        assertThat(context.forwarded().get(0).record().value().primaryPartition(), is(nullValue()));
+    }
+
     // Left join tests
     @Test
     public void leftJoinShouldPropagateNewPrimaryKeyWithNonNullFK() {

streams/src/test/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionWrapperSerdeTest.java

@@ -72,6 +72,30 @@ public void shouldSerdeV0Test() {
         assertEquals(version, deserialized.version());
     }
 
+    @Test
+    @SuppressWarnings("unchecked")
+    public void shouldSerdeV1WithNullPrimaryPartitionTest() {
+        final byte version = SubscriptionWrapper.VERSION_1;
+        final String originalKey = "originalKey";
+        final SubscriptionWrapperSerde swSerde = new SubscriptionWrapperSerde<>(() -> TOPIC, Serdes.String());
+        final long[] hashedValue = Murmur3.hash128(new byte[] {(byte) 0xFF, (byte) 0xAA, (byte) 0x00, (byte) 0x19});
+        final SubscriptionWrapper wrapper = new SubscriptionWrapper<>(
+                hashedValue,
+                SubscriptionWrapper.Instruction.DELETE_KEY_AND_PROPAGATE,
+                originalKey,
+                version,
+                null);
+        final byte[] serialized = swSerde.serializer().serialize(null, HEADERS, wrapper);
+        final SubscriptionWrapper deserialized = (SubscriptionWrapper) swSerde.deserializer()
+                .deserialize(null, HEADERS, serialized);
+
+        assertEquals(SubscriptionWrapper.Instruction.DELETE_KEY_AND_PROPAGATE, deserialized.instruction());
+        assertArrayEquals(hashedValue, deserialized.hash());
+        assertEquals(originalKey, deserialized.primaryKey());
+        assertNull(deserialized.primaryPartition());
+        assertEquals(version, deserialized.version());
+    }
+
     @Test
     @SuppressWarnings("unchecked")
     public void shouldSerdeV1Test() {
@@ -247,21 +271,6 @@ public void shouldThrowExceptionOnNullInstructionV1Test() {
             primaryPartition));
     }
 
-    @Test
-    public void shouldThrowExceptionOnNullPrimaryPartitionV1Test() {
-        final SubscriptionWrapperSerde swSerde = new SubscriptionWrapperSerde<>(() -> TOPIC, Serdes.String());
-        final String originalKey = "originalKey";
-        final long[] hashedValue = Murmur3.hash128(new byte[] {(byte) 0xFF, (byte) 0xAA, (byte) 0x00, (byte) 0x19});
-        final Integer primaryPartition = null;
-        final SubscriptionWrapper wrapper = new SubscriptionWrapper<>(
-            hashedValue,
-            SubscriptionWrapper.Instruction.PROPAGATE_ONLY_IF_FK_VAL_AVAILABLE,
-            originalKey,
-            SubscriptionWrapper.VERSION_1,
-            primaryPartition);
-        assertThrows(NullPointerException.class, () -> swSerde.serializer().serialize(null, HEADERS, wrapper));
-    }
-
     @Test
     public void shouldThrowExceptionOnUnsupportedVersionTest() {
         final String originalKey = "originalKey";