This is an automated email from the ASF dual-hosted git repository.
lucasbru 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 3e0787427b0 KAFKA-19871 - Add partition support to TestRecord (#22612)
3e0787427b0 is described below
commit 3e0787427b0500f216b306b6dc7927194bfa22f7
Author: Sebastien Viale <[email protected]>
AuthorDate: Tue Jun 23 16:05:46 2026 +0200
KAFKA-19871 - Add partition support to TestRecord (#22612)
**Motivation**
As part of [KIP-1238: Multi-partition support in
TopologyTestDriver.](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1238%3A+Multipartition+for+TopologyTestDriver+in+Kafka+Streams)
TestRecord needs to carry partition information so that records produced
and consumed by TopologyTestDriver can expose the partition they belong
to. This is a prerequisite for supporting multi-partition testing with
TopologyTestDriver.
**Changes**
This PR updates TestRecord by:
- adding a partition field;
- adding constructors that allow specifying the partition explicitly;
- preserving existing constructors, which continue to default the
partition to -1 (unspecified);
- including the partition in equals(), hashCode(), and toString();
- adding equalsIgnorePartition() for tests that do not care about the
partition.
**Compatibility**
Existing TestRecord constructors remain unchanged and continue to assign
partition = -1.
Since no component currently sets a partition value, records created
through existing APIs continue to carry partition = -1, preserving the
behavior of existing tests.
This PR is intentionally limited to introducing partition support in
TestRecord. Follow-up PRs will add multi-partition support to
TopologyTestDriver.
Reviewers: Matthias J. Sax <[email protected]>, Lucas Brutschy
<[email protected]>
Co-authored-by: Marie-Laure Momplot <[email protected]>
Co-authored-by: Julien Brunet <[email protected]>
Co-authored-by: Adam Souquieres <[email protected]>
---
.../org/apache/kafka/streams/test/TestRecord.java | 144 ++++++++++++++++-----
.../apache/kafka/streams/test/TestRecordTest.java | 128 +++++++++++++++++-
2 files changed, 238 insertions(+), 34 deletions(-)
diff --git
a/streams/test-utils/src/main/java/org/apache/kafka/streams/test/TestRecord.java
b/streams/test-utils/src/main/java/org/apache/kafka/streams/test/TestRecord.java
index eb724a08e16..1a6570a71e0 100644
---
a/streams/test-utils/src/main/java/org/apache/kafka/streams/test/TestRecord.java
+++
b/streams/test-utils/src/main/java/org/apache/kafka/streams/test/TestRecord.java
@@ -33,37 +33,62 @@ import java.util.StringJoiner;
* {@link TestInputTopic} will auto advance it's time when the record is piped.
*/
public class TestRecord<K, V> {
+ private static final int NO_PARTITION = -1;
private final Headers headers;
private final K key;
private final V value;
private final Instant recordTime;
-
- public boolean equalsIgnorePartition(final TestRecord<? extends K, ? super
V> o) {
- return false;
- }
+ /**
+ * The partition this record is assigned to.
+ * A value of {@code -1} is a sentinel meaning "no explicit partition set"
and is used
+ * only on <em>input</em> records created without an explicit partition
argument.
+ * Output records read from {@link
org.apache.kafka.streams.TestOutputTopic#readRecordsToList()}
+ * always carry the real resolved partition ({@code >= 0}).
+ */
+ private final int partition;
/**
- * Creates a record.
+ * Creates a record with an explicit partition.
*
* @param key The key that will be included in the record
* @param value The value of the record
- * @param headers the record headers that will be included in the record
- * @param recordTime The timestamp of the record.
+ * @param headers The record headers that will be included in the record
+ * @param recordTime The timestamp of the record
+ * @param partition The partition this record is assigned to
*/
- public TestRecord(final K key, final V value, final Headers headers, final
Instant recordTime) {
+ public TestRecord(final K key, final V value, final Headers headers, final
Instant recordTime, final int partition) {
this.key = key;
this.value = value;
this.recordTime = recordTime;
this.headers = new RecordHeaders(headers);
+ if (partition < NO_PARTITION) {
+ throw new IllegalArgumentException(
+ String.format("Invalid partition: %d. Partition number should
always be non-negative or %d.", partition, NO_PARTITION));
+ }
+ this.partition = partition;
}
/**
* Creates a record.
+ * Partition defaults to {@code -1} (no explicit partition set).
*
* @param key The key that will be included in the record
* @param value The value of the record
- * @param headers the record headers that will be included in the record
- * @param timestampMs The timestamp of the record, in milliseconds since
the beginning of the epoch.
+ * @param headers The record headers that will be included in the record
+ * @param recordTime The timestamp of the record
+ */
+ public TestRecord(final K key, final V value, final Headers headers, final
Instant recordTime) {
+ this(key, value, headers, recordTime, NO_PARTITION);
+ }
+
+ /**
+ * Creates a record.
+ * Partition defaults to {@code -1} (no explicit partition set).
+ *
+ * @param key The key that will be included in the record
+ * @param value The value of the record
+ * @param headers The record headers that will be included in the record
+ * @param timestampMs The timestamp of the record, in milliseconds since
the beginning of the epoch
*/
public TestRecord(final K key, final V value, final Headers headers, final
Long timestampMs) {
if (timestampMs != null) {
@@ -78,35 +103,36 @@ public class TestRecord<K, V> {
this.key = key;
this.value = value;
this.headers = new RecordHeaders(headers);
+ this.partition = NO_PARTITION;
}
/**
* Creates a record.
+ * Partition defaults to {@code -1} (no explicit partition set).
*
* @param key The key of the record
* @param value The value of the record
- * @param recordTime The timestamp of the record as Instant.
+ * @param recordTime The timestamp of the record as Instant
*/
public TestRecord(final K key, final V value, final Instant recordTime) {
- this(key, value, null, recordTime);
+ this(key, value, null, recordTime, NO_PARTITION);
}
/**
* Creates a record.
+ * Partition defaults to {@code -1} (no explicit partition set).
*
* @param key The key of the record
* @param value The value of the record
* @param headers The record headers that will be included in the record
*/
public TestRecord(final K key, final V value, final Headers headers) {
- this.key = key;
- this.value = value;
- this.headers = new RecordHeaders(headers);
- this.recordTime = null;
+ this(key, value, headers, (Instant) null, NO_PARTITION);
}
-
+
/**
* Creates a record.
+ * Partition defaults to {@code -1} (no explicit partition set).
*
* @param key The key of the record
* @param value The value of the record
@@ -116,10 +142,12 @@ public class TestRecord<K, V> {
this.value = value;
this.headers = new RecordHeaders();
this.recordTime = null;
+ this.partition = NO_PARTITION;
}
/**
* Create a record with {@code null} key.
+ * Partition defaults to {@code -1} (no explicit partition set).
*
* @param value The value of the record
*/
@@ -129,19 +157,27 @@ public class TestRecord<K, V> {
/**
* Create a {@code TestRecord} from a {@link ConsumerRecord}.
+ * The partition is taken from {@link ConsumerRecord#partition()}.
*
- * @param record The v
+ * @param record The consumer record
*/
public TestRecord(final ConsumerRecord<K, V> record) {
Objects.requireNonNull(record);
this.key = record.key();
this.value = record.value();
this.headers = record.headers();
- this.recordTime = Instant.ofEpochMilli(record.timestamp());
+ this.recordTime = record.timestamp() < 0 ? null :
Instant.ofEpochMilli(record.timestamp());
+ final int partition = record.partition();
+ if (partition < 0) {
+ throw new IllegalArgumentException(
+ String.format("Invalid partition: %d. Partition number
should always be non-negative.", partition));
+ }
+ this.partition = partition;
}
/**
* Create a {@code TestRecord} from a {@link ProducerRecord}.
+ * If the producer record carries an explicit partition it is used;
otherwise defaults to {@code -1}.
*
* @param record The record contents
*/
@@ -150,7 +186,18 @@ public class TestRecord<K, V> {
this.key = record.key();
this.value = record.value();
this.headers = record.headers();
- this.recordTime = Instant.ofEpochMilli(record.timestamp());
+ final Long timestamp = record.timestamp();
+ if (timestamp != null && timestamp < 0) {
+ throw new IllegalArgumentException(
+ String.format("Invalid timestamp: %d. Timestamp should always
be non-negative or null.", timestamp));
+ }
+ this.recordTime = timestamp == null ? null :
Instant.ofEpochMilli(timestamp);
+ final Integer partition = record.partition();
+ if (partition != null && partition < 0) {
+ throw new IllegalArgumentException(
+ String.format("Invalid partition: %d. Partition number should
always be non-negative or null.", partition));
+ }
+ this.partition = partition != null ? partition : NO_PARTITION;
}
/**
@@ -181,6 +228,13 @@ public class TestRecord<K, V> {
return this.recordTime == null ? null : this.recordTime.toEpochMilli();
}
+ /**
+ * @return the partition number, or {@code -1} if no partition was
explicitly set
+ */
+ public int partition() {
+ return partition;
+ }
+
/**
* @return The headers.
*/
@@ -209,17 +263,48 @@ public class TestRecord<K, V> {
return recordTime;
}
+ /**
+ * @return the partition number, or {@code -1} if no partition was
explicitly set
+ */
+ public int getPartition() {
+ return partition;
+ }
+
+ /**
+ * Compares this record to {@code otherRecord} without considering the
{@code partition} field.
+ *
+ * <p>Use this in tests that do not care about which partition a record
was routed to:
+ * <pre>{@code
+ * assertTrue(expected.equalsIgnorePartition(actual));
+ * }</pre>
+ *
+ * @param otherRecord the record to compare against; {@code null} returns
{@code false}
+ * @return {@code true} if all fields except {@code partition} are equal
+ */
+ public boolean equalsIgnorePartition(final TestRecord<K, V> otherRecord) {
+ return otherRecord != null && (this == otherRecord ||
equalsFields(otherRecord));
+ }
+
+ private boolean equalsFields(final TestRecord<K, V> otherRecord) {
+ return Objects.equals(headers, otherRecord.headers)
+ && Objects.equals(key, otherRecord.key)
+ && Objects.equals(value, otherRecord.value)
+ && Objects.equals(recordTime, otherRecord.recordTime);
+ }
+
@Override
public String toString() {
return new StringJoiner(", ", TestRecord.class.getSimpleName() + "[",
"]")
- .add("key=" + key)
- .add("value=" + value)
- .add("headers=" + headers)
- .add("recordTime=" + recordTime)
- .toString();
+ .add("key=" + key)
+ .add("value=" + value)
+ .add("headers=" + headers)
+ .add("recordTime=" + recordTime)
+ .add("partition=" + partition)
+ .toString();
}
@Override
+ @SuppressWarnings("unchecked")
public boolean equals(final Object o) {
if (this == o) {
return true;
@@ -227,15 +312,12 @@ public class TestRecord<K, V> {
if (o == null || getClass() != o.getClass()) {
return false;
}
- final TestRecord<?, ?> that = (TestRecord<?, ?>) o;
- return Objects.equals(headers, that.headers) &&
- Objects.equals(key, that.key) &&
- Objects.equals(value, that.value) &&
- Objects.equals(recordTime, that.recordTime);
+ final TestRecord<K, V> that = (TestRecord<K, V>) o;
+ return equalsFields(that) && partition == that.partition;
}
@Override
public int hashCode() {
- return Objects.hash(headers, key, value, recordTime);
+ return Objects.hash(headers, key, value, recordTime, partition);
}
}
diff --git
a/streams/test-utils/src/test/java/org/apache/kafka/streams/test/TestRecordTest.java
b/streams/test-utils/src/test/java/org/apache/kafka/streams/test/TestRecordTest.java
index ad3b1a2a20a..3bde0342fa4 100644
---
a/streams/test-utils/src/test/java/org/apache/kafka/streams/test/TestRecordTest.java
+++
b/streams/test-utils/src/test/java/org/apache/kafka/streams/test/TestRecordTest.java
@@ -34,8 +34,11 @@ import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.allOf;
import static org.hamcrest.Matchers.hasProperty;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
public class TestRecordTest {
private final String key = "testKey";
@@ -147,7 +150,7 @@ public class TestRecordTest {
assertThat(testRecord.toString(), equalTo("TestRecord[key=testKey,
value=1, "
+ "headers=RecordHeaders(headers = [RecordHeader(key = foo,
value = [118, 97, 108, 117, 101]), "
+ "RecordHeader(key = bar, value = null), RecordHeader(key =
\"A\\u00ea\\u00f1\\u00fcC\", value = [118, 97, 108, 117, 101])], isReadOnly =
false), "
- + "recordTime=2019-06-01T10:00:00Z]"));
+ + "recordTime=2019-06-01T10:00:00Z, partition=-1]"));
}
@Test
@@ -156,17 +159,136 @@ public class TestRecordTest {
final ConsumerRecord<String, Integer> consumerRecord = new
ConsumerRecord<>(topicName, 1, 0, recordMs,
TimestampType.CREATE_TIME, 0, 0, key, value, headers,
Optional.empty());
final TestRecord<String, Integer> testRecord = new
TestRecord<>(consumerRecord);
- final TestRecord<String, Integer> expectedRecord = new
TestRecord<>(key, value, headers, recordTime);
+ final TestRecord<String, Integer> expectedRecord = new
TestRecord<>(key, value, headers, recordTime, 1);
assertEquals(expectedRecord, testRecord);
}
+ @Test
+ public void testConsumerRecordWithNegativePartition() {
+ final String topicName = "topic";
+ final ConsumerRecord<String, Integer> consumerRecord = new
ConsumerRecord<>(topicName, -1, 0, recordMs,
+ TimestampType.CREATE_TIME, 0, 0, key, value, headers,
Optional.empty());
+ final IllegalArgumentException exception =
assertThrows(IllegalArgumentException.class, () -> new
TestRecord<>(consumerRecord));
+ assertEquals("Invalid partition: -1. Partition number should always be
non-negative.",
+ exception.getMessage());
+ }
+
+ @Test
+ public void testConsumerRecordWithNoTimestamp() {
+ final String topicName = "topic";
+ final ConsumerRecord<String, String> record = new ConsumerRecord<>(
+ topicName, 0, 0L, ConsumerRecord.NO_TIMESTAMP,
TimestampType.NO_TIMESTAMP_TYPE,
+ 0, 0, "key", "value", new RecordHeaders(), Optional.empty()
+ );
+ final TestRecord<String, String> testRecord = new TestRecord<>(record);
+ assertNull(testRecord.timestamp());
+ }
+
@Test
public void testProducerRecord() {
final String topicName = "topic";
final ProducerRecord<String, Integer> producerRecord =
new ProducerRecord<>(topicName, 1, recordMs, key, value, headers);
final TestRecord<String, Integer> testRecord = new
TestRecord<>(producerRecord);
- final TestRecord<String, Integer> expectedRecord = new
TestRecord<>(key, value, headers, recordTime);
+ final TestRecord<String, Integer> expectedRecord = new
TestRecord<>(key, value, headers, recordTime, 1);
assertEquals(expectedRecord, testRecord);
}
+
+ @Test
+ public void testProducerRecordWithNullTimestamp() {
+ final String topicName = "topic";
+ final ProducerRecord<String, String> record = new ProducerRecord<>(
+ topicName, null, null, "key", "value", new RecordHeaders()
+ );
+ final TestRecord<String, String> testRecord = new TestRecord<>(record);
+ assertNull(testRecord.timestamp());
+ }
+
+ @Test
+ public void testProducerRecordWithoutPartition() {
+ final String topicName = "topic";
+ final ProducerRecord<String, Integer> producerRecord =
+ new ProducerRecord<>(topicName, null, recordMs, key, value,
headers);
+ final TestRecord<String, Integer> testRecord = new
TestRecord<>(producerRecord);
+ assertEquals(-1, testRecord.partition());
+ }
+
+ @Test
+ public void testPartitionDefaultsToUnset() {
+ // Records built without an explicit partition, default to -1
+ assertEquals(-1, new TestRecord<>(key, value, headers,
recordTime).partition());
+ assertEquals(-1, new TestRecord<>(key, value, headers,
recordMs).partition());
+ assertEquals(-1, new TestRecord<>(key, value, headers).partition());
+ assertEquals(-1, new TestRecord<>(key, value).partition());
+ assertEquals(-1, new TestRecord<>(value).partition());
+ }
+
+ @Test
+ public void testExplicitPartitionConstructor() {
+ // Records built with an explicit partition.
+ final TestRecord<String, Integer> testRecord = new TestRecord<>(key,
value, headers, recordTime, 3);
+ assertEquals(3, testRecord.partition());
+ }
+
+ @Test
+ public void testExplicitNoPartitionConstructor() {
+ // Records built with the NO_PARTITION sentinel.
+ final TestRecord<String, Integer> testRecord =
+ new TestRecord<>(key, value, headers, recordTime, -1);
+ assertEquals(-1, testRecord.partition());
+ }
+
+ @Test
+ public void testInvalidNegativePartitionConstructor() {
+ final IllegalArgumentException exception = assertThrows(
+ IllegalArgumentException.class,
+ () -> new TestRecord<>(key, value, headers, recordTime, -2)
+ );
+ assertEquals(
+ "Invalid partition: -2. Partition number should always be
non-negative or -1.",
+ exception.getMessage()
+ );
+ }
+
+ @Test
+ public void testEqualsConsidersPartition() {
+ // equals()/hashCode() take the partition into account.
+ final TestRecord<String, Integer> record1 = new TestRecord<>(key,
value, headers, recordTime, 0);
+ final TestRecord<String, Integer> record2 = new TestRecord<>(key,
value, headers, recordTime, 1);
+ assertNotEquals(record1, record2);
+
+ final TestRecord<String, Integer> record1Again = new TestRecord<>(key,
value, headers, recordTime, 0);
+ assertEquals(record1, record1Again);
+ assertEquals(record1.hashCode(), record1Again.hashCode());
+
+ // an unset (default) partition differs from an explicit one
+ assertNotEquals(new TestRecord<>(key, value, headers, recordTime),
record1);
+ }
+
+ @Test
+ public void testEqualsIgnorePartition() {
+ // equalsIgnorePartition() matches on every field except the partition.
+ final TestRecord<String, Integer> record1 = new TestRecord<>(key,
value, headers, recordTime, 0);
+ final TestRecord<String, Integer> record2 = new TestRecord<>(key,
value, headers, recordTime, 1);
+ assertNotEquals(record1, record2);
+ assertTrue(record1.equalsIgnorePartition(record2));
+ assertTrue(record2.equalsIgnorePartition(record1));
+
+ // a genuine field mismatch is still detected
+ assertFalse(record1.equalsIgnorePartition(new TestRecord<>("other",
value, headers, recordTime, 0)));
+ assertFalse(record1.equalsIgnorePartition(new TestRecord<>(key, 2,
headers, recordTime, 0)));
+
+ // reflexive / null guards
+ assertTrue(record1.equalsIgnorePartition(record1));
+ assertFalse(record1.equalsIgnorePartition(null));
+ }
+
+ @Test
+ public void testToStringIncludesPartitionWhenSet() {
+ final TestRecord<String, Integer> testRecord = new TestRecord<>(key,
value, headers, recordTime, 2);
+ assertThat(testRecord.toString(), equalTo("TestRecord[key=testKey,
value=1, "
+ + "headers=RecordHeaders(headers = [RecordHeader(key = foo, value
= [118, 97, 108, 117, 101]), "
+ + "RecordHeader(key = bar, value = null), RecordHeader(key =
\"A\\u00ea\\u00f1\\u00fcC\", value = [118, 97, 108, 117, 101])], isReadOnly =
false), "
+ + "recordTime=2019-06-01T10:00:00Z, partition=2]"));
+ }
}
