KAFKA-20741: Add ReadOnlyRecord IQv2 result type (KIP-1356)

streams/src/main/java/org/apache/kafka/streams/processor/api/ReadOnlyRecord.java

@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.processor.api;
+
+import org.apache.kafka.common.header.Headers;
+
+/**
+ * A read-only view of a record's {@code key}, {@code value}, {@code timestamp}, and {@code headers}.
+ *
+ * <p>This is the shared read surface of a record: the processing-layer {@link Record} extends it with
+ * transform-and-forward builders ({@code withKey}/{@code withValue}/{@code withTimestamp}/{@code withHeaders}),
+ * while an Interactive Query (IQv2) result is exactly this read-only snapshot and exposes nothing more.
+ * It is the result type returned by the headers-aware IQv2 query types (e.g.
+ * {@code TimestampedKeyWithHeadersQuery}), which surface the record headers
+ * persisted by header-aware state stores.
+ *
+ * @param <K> The type of the key
+ * @param <V> The type of the value
+ */
+public interface ReadOnlyRecord<K, V> {
+
+    /**
+     * The key of the record. May be null.
+     */
+    K key();
+
+    /**
+     * The value of the record. May be null.
+     */
+    V value();
+
+    /**
+     * The timestamp of the record. Will never be negative.
+     */
+    long timestamp();
+
+    /**
+     * The headers of the record. Never null.
+     */
+    Headers headers();
+}

streams/src/main/java/org/apache/kafka/streams/processor/api/Record.java

@@ -35,7 +35,7 @@
  * @param <K> The type of the key
  * @param <V> The type of the value
  */
-public class Record<K, V> {
+public class Record<K, V> implements ReadOnlyRecord<K, V> {
     private final K key;
     private final V value;
     private final long timestamp;

streams/src/test/java/org/apache/kafka/streams/processor/api/ReadOnlyRecordTest.java

@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.processor.api;
+
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.header.internals.RecordHeader;
+import org.apache.kafka.common.header.internals.RecordHeaders;
+
+import org.junit.jupiter.api.Test;
+
+import java.nio.charset.StandardCharsets;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class ReadOnlyRecordTest {
+
+    private static Headers headers() {
+        return new RecordHeaders().add(new RecordHeader("k", "v".getBytes(StandardCharsets.UTF_8)));
+    }
+
+    @Test
+    public void recordShouldBeAssignableToReadOnlyRecordAndExposeSameAccessors() {
+        final Headers headers = headers();
+        final Record<String, String> record = new Record<>("key", "value", 123L, headers);
+
+        // The assignment itself proves Record is a ReadOnlyRecord (source/binary compatible change).
+        final ReadOnlyRecord<String, String> readOnly = record;
+
+        assertEquals(record.key(), readOnly.key());
+        assertEquals(record.value(), readOnly.value());
+        assertEquals(record.timestamp(), readOnly.timestamp());
+        assertEquals(record.headers(), readOnly.headers());
+
+        assertEquals("key", readOnly.key());
+        assertEquals("value", readOnly.value());
+        assertEquals(123L, readOnly.timestamp());
+        assertEquals(headers, readOnly.headers());
+    }
+
+    @Test
+    public void readOnlyHeadersShouldBeNonNullAndEmptyWhenConstructedWithoutHeaders() {
+        final ReadOnlyRecord<String, String> readOnly = new Record<>("key", "value", 123L);
+        assertEquals(new RecordHeaders(), readOnly.headers());
+    }
+
+    @Test
+    public void readOnlyAccessorsShouldTolerateNullKeyAndValue() {
+        final ReadOnlyRecord<String, String> readOnly = new Record<>(null, null, 0L);
+        assertEquals(null, readOnly.key());
+        assertEquals(null, readOnly.value());
+        assertEquals(0L, readOnly.timestamp());
+        assertEquals(new RecordHeaders(), readOnly.headers());
+    }
+}