lucasbru commented on code in PR #22612:
URL: https://github.com/apache/kafka/pull/22612#discussion_r3443309919
##########
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:
Maybe validate against negative values? Can I pass the sentinel here?
probably not
##########
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:
`ProducerRecord` throws on negative partitions, this one silently fixed.
Probably both should throw?
--
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]
