Copilot commented on code in PR #22612:
URL: https://github.com/apache/kafka/pull/22612#discussion_r3444083571
##########
streams/test-utils/src/main/java/org/apache/kafka/streams/test/TestRecord.java:
##########
@@ -33,37 +33,58 @@
* {@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);
+ this.partition = partition;
}
Review Comment:
The new (K,V,Headers,Instant,int) constructor accepts arbitrary negative
partition values (eg -2), which creates an invalid record state and is
inconsistent with the validation done in the ProducerRecord-based constructor.
Consider validating that the partition is either -1 (unset) or >= 0.
##########
streams/test-utils/src/main/java/org/apache/kafka/streams/test/TestRecord.java:
##########
@@ -138,10 +163,12 @@ public TestRecord(final ConsumerRecord<K, V> record) {
this.value = record.value();
this.headers = record.headers();
this.recordTime = Instant.ofEpochMilli(record.timestamp());
+ this.partition = record.partition() < 0 ? NO_PARTITION :
record.partition();
Review Comment:
ConsumerRecord can carry NO_TIMESTAMP (-1) when no timestamp is present.
Converting it via Instant.ofEpochMilli(-1) produces a bogus pre-epoch time
instead of preserving the 'no timestamp' semantics (null in TestRecord). It
would also be good to guard against any other unexpected negative timestamps.
##########
streams/test-utils/src/main/java/org/apache/kafka/streams/test/TestRecord.java:
##########
@@ -151,6 +178,13 @@ public TestRecord(final ProducerRecord<K, V> record) {
this.value = record.value();
this.headers = record.headers();
this.recordTime = Instant.ofEpochMilli(record.timestamp());
+ final Integer partition = record.partition();
+ if (partition != null && partition < 0) {
+ throw new IllegalArgumentException(
+ "Partition must be >= 0 or null, got: " + partition
+ );
+ }
+ this.partition = partition != null ? partition : NO_PARTITION;
Review Comment:
ProducerRecord#timestamp() is nullable. Calling
Instant.ofEpochMilli(record.timestamp()) will throw NPE when the ProducerRecord
was created without an explicit timestamp. TestRecord should preserve the
existing "no timestamp" semantics by setting recordTime to null in this case
(and optionally reject negative timestamps for consistency with the other
constructors).
##########
streams/test-utils/src/main/java/org/apache/kafka/streams/test/TestRecord.java:
##########
@@ -33,37 +33,58 @@
* {@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}).
Review Comment:
The field Javadoc states that output records from
TestOutputTopic#readRecordsToList() always carry a resolved partition (>= 0).
However, TopologyTestDriver#readRecord currently constructs TestRecord without
propagating any partition metadata, so this guarantee is not true yet. Please
adjust the wording to reflect current behavior (partition may be -1 until
partition propagation is implemented).
##########
streams/test-utils/src/main/java/org/apache/kafka/streams/test/TestRecord.java:
##########
@@ -129,6 +153,7 @@ public TestRecord(final V value) {
/**
* Create a {@code TestRecord} from a {@link ConsumerRecord}.
+ * The partition is taken from {@link ConsumerRecord#partition()}.
*
* @param record The v
*/
Review Comment:
The ConsumerRecord-based constructor Javadoc has an unhelpful parameter
description (`@param record The v`). Please update it to describe the parameter
properly.
##########
streams/test-utils/src/test/java/org/apache/kafka/streams/test/TestRecordTest.java:
##########
@@ -156,7 +158,7 @@ public void testConsumerRecord() {
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);
Review Comment:
There are no tests covering ProducerRecord created without an explicit
timestamp (timestamp=null) or ConsumerRecord with NO_TIMESTAMP (-1). Given the
constructors now handle partition metadata, adding small regression tests for
these timestamp edge cases would help ensure TestRecord continues to represent
"no timestamp" as null rather than throwing NPE or creating a pre-epoch Instant.
##########
streams/test-utils/src/main/java/org/apache/kafka/streams/test/TestRecord.java:
##########
@@ -209,14 +250,44 @@ public Instant getRecordTime() {
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) {
Review Comment:
equalsIgnorePartition previously existed with a wildcard signature
(TestRecord<? extends K, ? super V>). Narrowing it to TestRecord<K, V> is a
source-incompatible API change and will break existing call sites that relied
on the more flexible generic bounds. Please keep the original signature and
implement it with the new logic.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
