This is an automated email from the ASF dual-hosted git repository.
frankvicky pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new cf8b84f2c7e KAFKA-20741: Add ReadOnlyRecord IQv2 result type
(KIP-1356) (#22677)
cf8b84f2c7e is described below
commit cf8b84f2c7ef4c1fe9905ab73bbf7fdb7fc994b7
Author: Jess668 <[email protected]>
AuthorDate: Tue Jun 30 14:06:26 2026 -0400
KAFKA-20741: Add ReadOnlyRecord IQv2 result type (KIP-1356) (#22677)
Introduce a read-only view of a record (key/value/timestamp/headers);
prerequisite for the KIP-1356 headers-aware IQv2 query types.
- ReadOnlyRecord<K, V> (new, org.apache.kafka.streams.processor.api) — a
read-only view
exposing key() / value() / timestamp() / headers().
- Record now implements ReadOnlyRecord.
- ReadOnlyRecordTest — Record is assignable to ReadOnlyRecord and
exposes the same key/value/timestamp/headers; headers are non-null/empty
by default; null key/value tolerated.
Reviewers: Alieh Saeedi <[email protected]>, TengYao Chi
<[email protected]>
---
.../streams/processor/api/ReadOnlyRecord.java | 62 +++++++++++++++++++++
.../apache/kafka/streams/processor/api/Record.java | 6 +-
.../streams/processor/api/ReadOnlyRecordTest.java | 64 ++++++++++++++++++++++
3 files changed, 131 insertions(+), 1 deletion(-)
diff --git
a/streams/src/main/java/org/apache/kafka/streams/processor/api/ReadOnlyRecord.java
b/streams/src/main/java/org/apache/kafka/streams/processor/api/ReadOnlyRecord.java
new file mode 100644
index 00000000000..6d204a76510
--- /dev/null
+++
b/streams/src/main/java/org/apache/kafka/streams/processor/api/ReadOnlyRecord.java
@@ -0,0 +1,62 @@
+/*
+ * 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.annotation.InterfaceStability.Evolving;
+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, 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
+ */
+@Evolving
+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.
+ *
+ * <p>The returned {@link Headers} is part of a read-only view and must
not be mutated by
+ * callers.
+ */
+ // TODO (KIP-1356 follow-up): once the IQv2 query types land, records
served as IQv2
+ // results from a state store will have their headers frozen via
RecordHeaders.setReadOnly(),
+ // so that any attempt to mutate them throws IllegalStateException.
+ Headers headers();
+}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/processor/api/Record.java
b/streams/src/main/java/org/apache/kafka/streams/processor/api/Record.java
index 225b95fa400..37645d52640 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/api/Record.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/api/Record.java
@@ -35,7 +35,7 @@ import java.util.Objects;
* @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;
@@ -86,6 +86,7 @@ public class Record<K, V> {
/**
* The key of the record. May be null.
*/
+ @Override
public K key() {
return key;
}
@@ -93,6 +94,7 @@ public class Record<K, V> {
/**
* The value of the record. May be null.
*/
+ @Override
public V value() {
return value;
}
@@ -100,6 +102,7 @@ public class Record<K, V> {
/**
* The timestamp of the record. Will never be negative.
*/
+ @Override
public long timestamp() {
return timestamp;
}
@@ -107,6 +110,7 @@ public class Record<K, V> {
/**
* The headers of the record. Never null.
*/
+ @Override
public Headers headers() {
return headers;
}
diff --git
a/streams/src/test/java/org/apache/kafka/streams/processor/api/ReadOnlyRecordTest.java
b/streams/src/test/java/org/apache/kafka/streams/processor/api/ReadOnlyRecordTest.java
new file mode 100644
index 00000000000..ee946119c39
--- /dev/null
+++
b/streams/src/test/java/org/apache/kafka/streams/processor/api/ReadOnlyRecordTest.java
@@ -0,0 +1,64 @@
+/*
+ * 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;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+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("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);
+ assertNull(readOnly.key());
+ assertNull(readOnly.value());
+ assertEquals(0L, readOnly.timestamp());
+ assertEquals(new RecordHeaders(), readOnly.headers());
+ }
+}
