mjsax commented on code in PR #22612:
URL: https://github.com/apache/kafka/pull/22612#discussion_r3439174299
##########
streams/test-utils/src/main/java/org/apache/kafka/streams/test/TestRecord.java:
##########
@@ -209,14 +243,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 o} without considering the {@code
partition} field.
Review Comment:
```suggestion
* Compares this record to {@code otherRecord} without considering the
{@code partition} field.
```
##########
streams/test-utils/src/main/java/org/apache/kafka/streams/test/TestRecord.java:
##########
@@ -138,10 +162,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();
Review Comment:
Technically, there is no guard to what value `ConsumerRecord#partition` is
set -- of course, if we get it from a `KafkaConsumer` we can expect it to be
correctly set. However, `new ConsumerRecord` is a public API so users could
create one manually. Wondering if we should have a guard to check for `< 0`,
and throw(?) or overwrite to `-1 `(?) for this case, to avoid weird corner
cases?
Or would we want to accept that we either crash in `< -1` or treat all
negative values as "not set" (ie, same as `-1`)?
##########
streams/test-utils/src/main/java/org/apache/kafka/streams/test/TestRecord.java:
##########
@@ -33,37 +33,57 @@
* {@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;
}
/**
* Creates a record.
*
* @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).
Review Comment:
Should this comment be also on the above constructor?
##########
streams/test-utils/src/main/java/org/apache/kafka/streams/test/TestRecord.java:
##########
@@ -151,6 +177,7 @@ public TestRecord(final ProducerRecord<K, V> record) {
this.value = record.value();
this.headers = record.headers();
this.recordTime = Instant.ofEpochMilli(record.timestamp());
+ this.partition = record.partition() != null ? record.partition() :
NO_PARTITION;
Review Comment:
Same question -- if `record.partition() != null`, it could have any value...
For producer value. For the producer case, treating anything negative as
invalid and throw could make even more sense compared to `ConsumerRecord`
because here we have the `null` option.
Thoughts?
##########
streams/test-utils/src/main/java/org/apache/kafka/streams/test/TestRecord.java:
##########
@@ -209,14 +243,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 o} 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 o 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<? extends K, ?
extends V> o) {
Review Comment:
Should it be only `TestRecord<K, V>` -- during KIP review, I was also
briefly thinking if we would want to support sub-types, but in the end, if we
compare for equality, a TestRecord with sub-key-type or sub-value-type is not
equals to "this" record, so we don't need to support this and can just use
`<K,V>` and disallow at compile time already?
##########
streams/test-utils/src/main/java/org/apache/kafka/streams/test/TestRecord.java:
##########
@@ -209,14 +243,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 o} 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 o 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<? extends K, ?
extends V> o) {
+ return o != null && (this == o || equalsFields(o));
+ }
+
+ private boolean equalsFields(final TestRecord<?, ?> other) {
Review Comment:
```suggestion
private boolean equalsFields(final TestRecord<K, V> otherRecord) {
```
##########
streams/test-utils/src/main/java/org/apache/kafka/streams/test/TestRecord.java:
##########
@@ -209,14 +243,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 o} 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 o the record to compare against; {@code null} returns {@code
false}
Review Comment:
```suggestion
* @param otherRecord the record to compare against; {@code null}
returns {@code false}
```
##########
streams/test-utils/src/main/java/org/apache/kafka/streams/test/TestRecord.java:
##########
@@ -209,14 +243,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 o} 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 o 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<? extends K, ?
extends V> o) {
Review Comment:
```suggestion
public boolean equalsIgnorePartition(final TestRecord<? extends K, ?
extends V> otherRecord) {
```
--
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]
