KAFKA-20413: Make the stream-stream outer join use the headers-aware ListValueStore [2/3]

streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/OuterJoinListValueStoreRestorationTest.java

@@ -17,7 +17,13 @@
 package org.apache.kafka.streams.integration;
 
 import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.IsolationLevel;
+import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.header.internals.RecordHeaders;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
 import org.apache.kafka.common.serialization.Serdes;
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.apache.kafka.common.serialization.StringSerializer;
@@ -42,9 +48,13 @@
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.MethodSource;
+import org.junit.jupiter.params.provider.ValueSource;
 
+import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.util.List;
+import java.util.Map;
+import java.util.Objects;
 import java.util.Properties;
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -55,8 +65,10 @@
 import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.startApplicationAndWaitUntilRunning;
 import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.waitForCompletion;
 import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived;
+import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.waitUntilMinRecordsReceived;
 import static org.apache.kafka.streams.utils.TestUtils.safeUniqueTestName;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 /**
  * Integration test for verifying ListValueStore deserialization behavior after state restoration
@@ -69,6 +81,16 @@ public class OuterJoinListValueStoreRestorationTest {
     private static final int NUM_BROKERS = 1;
     public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(NUM_BROKERS);
 
+    /** Stands in for an absent record header, so a lost header shows up as a diff rather than an NPE. */
+    private static final String NO_HEADER = "<no header>";
+
+    /**
+     * {@code ListValueStoreUpgradeUtils.LIST_VALUE_HEADERS_HEADER_KEY}, repeated as a literal because that
+     * constant is package-private in {@code org.apache.kafka.streams.state.internals}. Only the
+     * headers-format changelogger writes it, which makes it the marker that HEADERS mode really took hold.
+     */
+    private static final String LIST_VALUE_HEADERS_HEADER_KEY = "vh";
+
     private String applicationId;
     private String leftTopic;
     private String rightTopic;
@@ -228,6 +250,206 @@ public void testOuterJoinRestorationWithMultipleRecords(final String processingG
             "No unexpected keys should appear on the output topic after restoration");
     }
 
+    /**
+     * Verifies that per-record headers stored in the header-aware outer-join store survive a full
+     * fault-tolerance cycle: the headers are written into the changelog, the local state is wiped,
+     * and after the store is rebuilt from the changelog the non-joined records are emitted carrying
+     * their original headers.
+     */
+    @ParameterizedTest
+    @ValueSource(strings = {StreamsConfig.AT_LEAST_ONCE, StreamsConfig.EXACTLY_ONCE_V2})
+    public void testOuterJoinHeadersSurviveRestoration(final String processingGuarantee) throws Exception {
+        // Headers are only stored in the header-aware store format.
+        streamsConfig.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, processingGuarantee);
+        streamsConfig.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
+
+        // Step 1: Initial Topology Start
+        streams = createOuterJoinTopology();
+        startApplicationAndWaitUntilRunning(streams);
+
+        // Step 2: Produce non-joined left records, each carrying a distinct header.
+        // CRITICAL: Do NOT advance the window yet, so the records (and their headers) stay in the store.
+        final Map<String, String> expectedHeaders = IntStream.range(0, 10).boxed()
+            .collect(Collectors.toMap(i -> "key" + i, i -> "v" + i));
+        long timestamp = 1000L;
+        for (int i = 0; i < 10; i++) {
+            final Headers headers = new RecordHeaders().add("h", ("v" + i).getBytes(StandardCharsets.UTF_8));
+            produceRecordWithHeaders(leftTopic, "key" + i, "left-" + i, headers, timestamp);
+            timestamp += 100;
+        }
+
+        // Wait for processing and commit to the changelog, proven via a probe join result.
+        produceRecord(leftTopic, "probe", "probe-left", timestamp);
+        produceRecord(rightTopic, "probe", "probe-right", timestamp);
+        waitUntilMinKeyValueRecordsReceived(getConsumerConfig(), outputTopic, 1, 30000);
+        waitForCompletion(streams, 2, 30000);
+
+        // Step 3: Force State Restoration (wipe local state, rebuild from changelog).
+        streams.close(Duration.ofSeconds(30));
+        purgeLocalStreamsState(streamsConfig);
+
+        // Step 4: Restart with Restoration
+        streams = createOuterJoinTopology();
+        startApplicationAndWaitUntilRunning(streams);
+
+        // Step 5: Advance the window to trigger emitNonJoinedOuterRecords() for the restored records.
+        produceRecord(leftTopic, "trigger", "trigger-value", 62000L); // beyond the 60-second window
+
+        final List<ConsumerRecord<String, String>> results =
+            waitUntilMinRecordsReceived(getConsumerConfig(), outputTopic, 10, 30000);
+
+        // Step 6: Each restored non-joined left record must be emitted with its original header.
+        // A missing header maps to a sentinel rather than null: that is the regression this test guards,
+        // and Collectors.toMap rejects null values, so it would surface as a bare NPE here instead of an
+        // assertEquals diff showing which keys lost their header.
+        final Map<String, String> actualHeaders = results.stream()
+            .filter(r -> r.value() != null && r.value().endsWith("right=null"))
+            .filter(r -> r.key().startsWith("key"))
+            .collect(Collectors.toMap(
+                ConsumerRecord::key,
+                r -> Objects.requireNonNullElse(headerValue(r, "h"), NO_HEADER),
+                (a, b) -> a));
+
+        assertEquals(expectedHeaders, actualHeaders,
+            "Each non-joined left record should retain its original header after wipe + changelog restoration");
+    }
+
+    /**
+     * The downgrade direction: state is written by a HEADERS-format store and then restored by a PLAIN
+     * one. This is what the changelog encoding exists for. If the local
+     * {@code [headersSize][headers][flag][value]} element format reached the changelog, the PLAIN reader
+     * would consume each element's leading empty-headers {@code 0x00} as the {@code LeftOrRightValue}
+     * flag and silently emit left records as right ones -- i.e. {@code (null, x)} instead of
+     * {@code (x, null)} -- with no exception to signal it.
+     * <p>
+     * Note this needs no version downgrade: flipping {@code dsl.store.format} back to PLAIN on one
+     * version is enough, and it is also the path the "delete the local state and rebuild the store from
+     * the changelog" advice in the {@code RocksDBStore} downgrade guard takes.
+     */
+    @ParameterizedTest
+    @ValueSource(strings = {StreamsConfig.AT_LEAST_ONCE, StreamsConfig.EXACTLY_ONCE_V2})
+    public void testHeadersStoreChangelogIsReadableByAPlainStore(final String processingGuarantee) throws Exception {
+        streamsConfig.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, processingGuarantee);
+        streamsConfig.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
+
+        // Step 1: run in HEADERS mode and park non-joined left records (with headers) in the store.
+        streams = createOuterJoinTopology();
+        startApplicationAndWaitUntilRunning(streams);
+
+        long timestamp = 1000L;
+        for (int i = 0; i < 10; i++) {
+            final Headers headers = new RecordHeaders().add("h", ("v" + i).getBytes(StandardCharsets.UTF_8));
+            produceRecordWithHeaders(leftTopic, "key" + i, "left-" + i, headers, timestamp);
+            timestamp += 100;
+        }
+
+        produceRecord(leftTopic, "probe", "probe-left", timestamp);
+        produceRecord(rightTopic, "probe", "probe-right", timestamp);
+        waitUntilMinKeyValueRecordsReceived(getConsumerConfig(), outputTopic, 1, 30000);
+        waitForCompletion(streams, 2, 30000);
+
+        // Step 1b: prove HEADERS mode actually took hold before relying on the downgrade below. Without
+        // this the test is vacuous: if the outer-join store ignored dsl.store.format, step 1 would log
+        // PLAIN elements and step 3 would read them back happily, so every assertion below would pass for
+        // the wrong reason -- including with this whole feature reverted.
+        assertTrue(
+            outerJoinChangelogCarriesElementHeaders(),
+            "No changelog record carried the " + LIST_VALUE_HEADERS_HEADER_KEY + " header, so the store was "
+                + "not the headers-format one and there is nothing for the downgrade below to prove");
+
+        // Step 2: wipe the local state, so the only surviving copy is the changelog.
+        streams.close(Duration.ofSeconds(30));
+        purgeLocalStreamsState(streamsConfig);
+
+        // Step 3: come back as a PLAIN store -- the downgrade.
+        streamsConfig.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_DEFAULT);
+        streams = createOuterJoinTopology();
+        startApplicationAndWaitUntilRunning(streams);
+
+        // Step 4: advance the window to emit the restored non-joined records.
+        produceRecord(leftTopic, "trigger", "trigger-value", 62000L);
+
+        final List<KeyValue<String, String>> results =
+            waitUntilMinKeyValueRecordsReceived(getConsumerConfig(), outputTopic, 10, 30000);
+
+        final Set<String> expectedKeys = IntStream.range(0, 10)
+            .mapToObj(i -> "key" + i)
+            .collect(Collectors.toSet());
+
+        // The join side must survive: every restored record is a LEFT record, so it is emitted as
+        // "left=<value>, right=null". Under the old changelog format these came back as right records.
+        final Set<String> emittedAsLeft = results.stream()
+            .filter(kv -> kv.value != null && kv.value.endsWith("right=null"))
+            .map(kv -> kv.key)
+            .collect(Collectors.toSet());
+        assertEquals(expectedKeys, emittedAsLeft,
+            "A PLAIN store restoring a HEADERS-written changelog must still see these as left records");
+
+        // The values must survive too: the old format left a stray NUL prepended to each value.
+        final Map<String, String> emittedValues = results.stream()
+            .filter(kv -> kv.key.startsWith("key"))
+            .collect(Collectors.toMap(kv -> kv.key, kv -> kv.value, (a, b) -> a));
+        final Map<String, String> expectedValues = IntStream.range(0, 10).boxed()
+            .collect(Collectors.toMap(i -> "key" + i, i -> "left=left-" + i + ", right=null"));
+        assertEquals(expectedValues, emittedValues,
+            "Values must be byte-identical after the downgrade, with no stray prefix");
+    }
+
+    private void produceRecordWithHeaders(final String topic,
+                                          final String key,
+                                          final String value,
+                                          final Headers headers,
+                                          final long timestamp) {
+        final Properties producerConfig = new Properties();
+        producerConfig.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
+        producerConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+        producerConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+        producerConfig.put(ProducerConfig.ACKS_CONFIG, "all");
+
+        IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+            topic,
+            List.of(new KeyValue<>(key, value)),
+            producerConfig,
+            headers,
+            timestamp,
+            false
+        );
+    }
+
+    /**
+     * Whether the outer-join store's changelog carries the per-element headers control header, i.e.
+     * whether the store writing it was the headers-format one.
+     */
+    private boolean outerJoinChangelogCarriesElementHeaders() throws Exception {
+        final Set<String> topics = CLUSTER.getAllTopicsInCluster();
+        // The topology does not name the join store, so it is KSTREAM-OUTERSHARED-<n>-store.
+        final String changelogTopic = topics.stream()
+            .filter(t -> t.contains("OUTERSHARED") && t.endsWith("-store-changelog"))
+            .findFirst()
+            .orElseThrow(() -> new AssertionError("no outer-join changelog topic among " + topics));
+
+        final Properties consumerConfig = new Properties();
+        consumerConfig.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
+        consumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, "changelog-consumer-" + applicationId);
+        consumerConfig.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
+        consumerConfig.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
+        consumerConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+        // Under exactly-once the changelog writes are transactional, so an uncommitted read could see
+        // records from an aborted transaction -- or miss the committed ones entirely.
+        consumerConfig.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, IsolationLevel.READ_COMMITTED.toString());
+
+        final List<ConsumerRecord<byte[], byte[]>> changelogRecords =
+            waitUntilMinRecordsReceived(consumerConfig, changelogTopic, 1, 30000);
+
+        return changelogRecords.stream()
+            .anyMatch(r -> r.headers().lastHeader(LIST_VALUE_HEADERS_HEADER_KEY) != null);
+    }
+
+    private static String headerValue(final ConsumerRecord<?, ?> record, final String key) {
+        final Header header = record.headers().lastHeader(key);
+        return header == null ? null : new String(header.value(), StandardCharsets.UTF_8);
+    }
+
     private void produceRecord(final String topic, final String key, final String value, final long timestamp) {
         final Properties producerConfig = new Properties();
         producerConfig.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());