showuon commented on code in PR #12753:
URL: https://github.com/apache/kafka/pull/12753#discussion_r1027448552
##########
core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala:
##########
@@ -694,6 +694,77 @@ class PlaintextConsumerTest extends BaseConsumerTest {
assertEquals(outOfRangePos.toLong, outOfRangePartitions.get(tp))
}
+ @Test
+ def testFetchInvalidOffsetResetConfigEarliest(): Unit = {
+ this.consumerConfig.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
"earliest")
+ val consumer = createConsumer(configOverrides = this.consumerConfig)
+ val totalRecords = 10L
+
+ val producer = createProducer()
+ val startingTimestamp = 0
+ sendRecords(producer, totalRecords.toInt, tp, startingTimestamp =
startingTimestamp)
+ consumer.assign(List(tp).asJava)
+ consumeAndVerifyRecords(consumer = consumer, numRecords =
totalRecords.toInt, startingOffset =0)
+ // seek to out of range position
+ val outOfRangePos = totalRecords + 1
+ consumer.seek(tp, outOfRangePos)
+ // assert that poll resets to the beginning position
+ consumeAndVerifyRecords(consumer = consumer, numRecords = 1,
startingOffset = 0)
+ }
+
+ @Test
+ def testFetchInvalidOffsetResetConfigLatestRecordsBeforeTimeout(): Unit = {
+
+ this.consumerConfig.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
"latest")
+ // long wait time should (theoretically) mean that more records show up
before this expires
+ this.consumerConfig.setProperty(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG,
"3000")
+ val consumer = createConsumer(configOverrides = this.consumerConfig)
+ val totalRecords = 10L
+
+ val producer = createProducer()
+ val startingTimestamp = 0
+ sendRecords(producer, totalRecords.toInt, tp, startingTimestamp =
startingTimestamp)
+ consumer.assign(List(tp).asJava)
+ consumer.seek(tp,0)
+ consumeAndVerifyRecords(consumer = consumer, numRecords =
totalRecords.toInt, startingOffset = 0)
+ // seek to out of range position
+ val outOfRangePos = totalRecords + 7 //arbitrary, much higher offset
+ consumer.seek(tp, outOfRangePos)
+ // assert that poll resets to the ending position
+ assertTrue(consumer.poll(Duration.ofMillis(50)).isEmpty)
+ //some new records show up before fetch max wait time
+ sendRecords(producer, totalRecords.toInt, tp, startingTimestamp =
totalRecords)
+ // ensure that new records start with the offset that was passed to
seek(), as new records have arrived
+ val nextRecord = consumer.poll(Duration.ofMillis(50)).iterator().next()
+ assertEquals(outOfRangePos,nextRecord.offset())
+
+ }
+
+ @Test
+ def testFetchInvalidOffsetResetConfigLatestNoRecordsBeforeTimeout(): Unit = {
Review Comment:
If you agree to remove the above test, this test could be simply renamed to
`testFetchOutOfRangeOffsetResetConfigLatest()`. WDYT?
##########
core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala:
##########
@@ -694,6 +694,77 @@ class PlaintextConsumerTest extends BaseConsumerTest {
assertEquals(outOfRangePos.toLong, outOfRangePartitions.get(tp))
}
+ @Test
+ def testFetchInvalidOffsetResetConfigEarliest(): Unit = {
+ this.consumerConfig.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
"earliest")
+ val consumer = createConsumer(configOverrides = this.consumerConfig)
+ val totalRecords = 10L
+
+ val producer = createProducer()
+ val startingTimestamp = 0
+ sendRecords(producer, totalRecords.toInt, tp, startingTimestamp =
startingTimestamp)
+ consumer.assign(List(tp).asJava)
+ consumeAndVerifyRecords(consumer = consumer, numRecords =
totalRecords.toInt, startingOffset =0)
+ // seek to out of range position
+ val outOfRangePos = totalRecords + 1
+ consumer.seek(tp, outOfRangePos)
+ // assert that poll resets to the beginning position
+ consumeAndVerifyRecords(consumer = consumer, numRecords = 1,
startingOffset = 0)
+ }
+
+ @Test
+ def testFetchInvalidOffsetResetConfigLatestRecordsBeforeTimeout(): Unit = {
Review Comment:
I don't think this test is testing reset offset logic. Could we remove this
test?
##########
core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala:
##########
@@ -694,6 +694,77 @@ class PlaintextConsumerTest extends BaseConsumerTest {
assertEquals(outOfRangePos.toLong, outOfRangePartitions.get(tp))
}
+ @Test
+ def testFetchInvalidOffsetResetConfigEarliest(): Unit = {
+ this.consumerConfig.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
"earliest")
+ val consumer = createConsumer(configOverrides = this.consumerConfig)
Review Comment:
Although everything works well now, I think we should still add
`this.consumerConfig.setProperty(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, "0")`
for earliest test, to make sure the offset reset is working as expected
##########
core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala:
##########
@@ -694,6 +694,77 @@ class PlaintextConsumerTest extends BaseConsumerTest {
assertEquals(outOfRangePos.toLong, outOfRangePartitions.get(tp))
}
+ @Test
+ def testFetchInvalidOffsetResetConfigEarliest(): Unit = {
+ this.consumerConfig.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
"earliest")
+ val consumer = createConsumer(configOverrides = this.consumerConfig)
+ val totalRecords = 10L
+
+ val producer = createProducer()
+ val startingTimestamp = 0
+ sendRecords(producer, totalRecords.toInt, tp, startingTimestamp =
startingTimestamp)
+ consumer.assign(List(tp).asJava)
+ consumeAndVerifyRecords(consumer = consumer, numRecords =
totalRecords.toInt, startingOffset =0)
+ // seek to out of range position
+ val outOfRangePos = totalRecords + 1
+ consumer.seek(tp, outOfRangePos)
+ // assert that poll resets to the beginning position
+ consumeAndVerifyRecords(consumer = consumer, numRecords = 1,
startingOffset = 0)
+ }
+
+ @Test
+ def testFetchInvalidOffsetResetConfigLatestRecordsBeforeTimeout(): Unit = {
+
+ this.consumerConfig.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
"latest")
+ // long wait time should (theoretically) mean that more records show up
before this expires
+ this.consumerConfig.setProperty(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG,
"3000")
+ val consumer = createConsumer(configOverrides = this.consumerConfig)
+ val totalRecords = 10L
+
+ val producer = createProducer()
+ val startingTimestamp = 0
+ sendRecords(producer, totalRecords.toInt, tp, startingTimestamp =
startingTimestamp)
+ consumer.assign(List(tp).asJava)
+ consumer.seek(tp,0)
+ consumeAndVerifyRecords(consumer = consumer, numRecords =
totalRecords.toInt, startingOffset = 0)
+ // seek to out of range position
+ val outOfRangePos = totalRecords + 7 //arbitrary, much higher offset
+ consumer.seek(tp, outOfRangePos)
+ // assert that poll resets to the ending position
+ assertTrue(consumer.poll(Duration.ofMillis(50)).isEmpty)
+ //some new records show up before fetch max wait time
+ sendRecords(producer, totalRecords.toInt, tp, startingTimestamp =
totalRecords)
+ // ensure that new records start with the offset that was passed to
seek(), as new records have arrived
+ val nextRecord = consumer.poll(Duration.ofMillis(50)).iterator().next()
+ assertEquals(outOfRangePos,nextRecord.offset())
+
+ }
+
+ @Test
+ def testFetchInvalidOffsetResetConfigLatestNoRecordsBeforeTimeout(): Unit = {
+ this.consumerConfig.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
"latest")
+ this.consumerConfig.setProperty(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG,
"0")
+ val consumer = createConsumer(configOverrides = this.consumerConfig)
+ val totalRecords = 10L
+
+ val producer = createProducer()
+ val startingTimestamp = 0
+ sendRecords(producer, totalRecords.toInt, tp, startingTimestamp =
startingTimestamp)
+ consumer.assign(List(tp).asJava)
+ consumer.seek(tp, 0)
+ //consume some, but not all of the records
+ consumeAndVerifyRecords(consumer = consumer, numRecords =
totalRecords.toInt/2, startingOffset = 0)
+ // seek to out of range position
+ val outOfRangePos = totalRecords + 17 //arbitrary, much higher offset
+ consumer.seek(tp, outOfRangePos)
+ // assert that poll resets to the ending position
+ assertTrue(consumer.poll(Duration.ofMillis(50)).isEmpty)
+ sendRecords(producer, totalRecords.toInt, tp, startingTimestamp =
totalRecords)
+ val nextRecord = consumer.poll(Duration.ofMillis(50)).iterator().next()
+ //ensure the seek went to the last known record at the time of the
previous poll
+ assertEquals(totalRecords,nextRecord.offset())
Review Comment:
Please add a space after comma, i.e. `assertEquals(totalRecords,
nextRecord.offset())`
##########
clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java:
##########
@@ -1581,6 +1581,23 @@ public void commitAsync(final Map<TopicPartition,
OffsetAndMetadata> offsets, Of
* is invoked for the same partition more than once, the latest offset
will be used on the next poll(). Note that
* you may lose data if this API is arbitrarily used in the middle of
consumption, to reset the fetch offsets
*
+ * The next Consumer Record which will be retrieved when poll() is invoked
will either have the offset specified or,
+ * by default, with {@link ConsumerConfig AUTO_RESET_CONFIG} set to
"latest"
+ * a higher numbered offset. A higher numbered offset will be returned if
there is no consumer record with the offset
+ * specified but there is one with a higher offset.
+ * seek(0) is equivalent to seek to beginning for a topic with beginning
offset 0.
Review Comment:
This is not always correct. The start offset might not be 0 if it got
deleted manually or exceeding retention time/size.... etc
* {@link #seekToBeginning(Collection)} is equivalent to seek to beginning ...
##########
clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java:
##########
@@ -1581,6 +1581,23 @@ public void commitAsync(final Map<TopicPartition,
OffsetAndMetadata> offsets, Of
* is invoked for the same partition more than once, the latest offset
will be used on the next poll(). Note that
* you may lose data if this API is arbitrarily used in the middle of
consumption, to reset the fetch offsets
*
+ * The next Consumer Record which will be retrieved when poll() is invoked
will either have the offset specified or,
+ * by default, with {@link ConsumerConfig AUTO_RESET_CONFIG} set to
"latest"
+ * a higher numbered offset. A higher numbered offset will be returned if
there is no consumer record with the offset
+ * specified but there is one with a higher offset.
+ * seek(0) is equivalent to seek to beginning for a topic with beginning
offset 0.
+ * seekToEnd() is equivalent to seeking to the highest known offset + 1.
+ *
+ * Seeking past the end of the highest known offset means an invalid
offset is reached.
+ * Invalid offset behaviour is controlled by
+ * the {@link ConsumerConfig AUTO_RESET_CONFIG} property. If this is set
to "earliest", the next poll() will seek to the beginning
+ * before returning a record. If it is set to "latest", it will seek to
the last known record (similar to seekToEnd()).
Review Comment:
seek will happen right after you call `seek`, it doesn't happen after
calling `poll`. How about this:
`If this is set to "earliest", the next poll will return records from the
starting offset. `
##########
core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala:
##########
@@ -694,6 +694,77 @@ class PlaintextConsumerTest extends BaseConsumerTest {
assertEquals(outOfRangePos.toLong, outOfRangePartitions.get(tp))
}
+ @Test
+ def testFetchInvalidOffsetResetConfigEarliest(): Unit = {
+ this.consumerConfig.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
"earliest")
+ val consumer = createConsumer(configOverrides = this.consumerConfig)
+ val totalRecords = 10L
+
+ val producer = createProducer()
+ val startingTimestamp = 0
+ sendRecords(producer, totalRecords.toInt, tp, startingTimestamp =
startingTimestamp)
+ consumer.assign(List(tp).asJava)
+ consumeAndVerifyRecords(consumer = consumer, numRecords =
totalRecords.toInt, startingOffset =0)
+ // seek to out of range position
+ val outOfRangePos = totalRecords + 1
+ consumer.seek(tp, outOfRangePos)
+ // assert that poll resets to the beginning position
+ consumeAndVerifyRecords(consumer = consumer, numRecords = 1,
startingOffset = 0)
+ }
+
+ @Test
+ def testFetchInvalidOffsetResetConfigLatestRecordsBeforeTimeout(): Unit = {
+
+ this.consumerConfig.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
"latest")
+ // long wait time should (theoretically) mean that more records show up
before this expires
+ this.consumerConfig.setProperty(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG,
"3000")
+ val consumer = createConsumer(configOverrides = this.consumerConfig)
+ val totalRecords = 10L
+
+ val producer = createProducer()
+ val startingTimestamp = 0
+ sendRecords(producer, totalRecords.toInt, tp, startingTimestamp =
startingTimestamp)
+ consumer.assign(List(tp).asJava)
+ consumer.seek(tp,0)
+ consumeAndVerifyRecords(consumer = consumer, numRecords =
totalRecords.toInt, startingOffset = 0)
+ // seek to out of range position
+ val outOfRangePos = totalRecords + 7 //arbitrary, much higher offset
+ consumer.seek(tp, outOfRangePos)
+ // assert that poll resets to the ending position
+ assertTrue(consumer.poll(Duration.ofMillis(50)).isEmpty)
+ //some new records show up before fetch max wait time
+ sendRecords(producer, totalRecords.toInt, tp, startingTimestamp =
totalRecords)
+ // ensure that new records start with the offset that was passed to
seek(), as new records have arrived
+ val nextRecord = consumer.poll(Duration.ofMillis(50)).iterator().next()
+ assertEquals(outOfRangePos,nextRecord.offset())
+
+ }
+
+ @Test
+ def testFetchInvalidOffsetResetConfigLatestNoRecordsBeforeTimeout(): Unit = {
+ this.consumerConfig.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
"latest")
+ this.consumerConfig.setProperty(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG,
"0")
+ val consumer = createConsumer(configOverrides = this.consumerConfig)
+ val totalRecords = 10L
+
+ val producer = createProducer()
+ val startingTimestamp = 0
+ sendRecords(producer, totalRecords.toInt, tp, startingTimestamp =
startingTimestamp)
+ consumer.assign(List(tp).asJava)
+ consumer.seek(tp, 0)
+ //consume some, but not all of the records
+ consumeAndVerifyRecords(consumer = consumer, numRecords =
totalRecords.toInt/2, startingOffset = 0)
+ // seek to out of range position
+ val outOfRangePos = totalRecords + 17 //arbitrary, much higher offset
Review Comment:
Please add a space after `//`
##########
core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala:
##########
@@ -694,6 +694,77 @@ class PlaintextConsumerTest extends BaseConsumerTest {
assertEquals(outOfRangePos.toLong, outOfRangePartitions.get(tp))
}
+ @Test
+ def testFetchInvalidOffsetResetConfigEarliest(): Unit = {
+ this.consumerConfig.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
"earliest")
+ val consumer = createConsumer(configOverrides = this.consumerConfig)
+ val totalRecords = 10L
+
+ val producer = createProducer()
+ val startingTimestamp = 0
+ sendRecords(producer, totalRecords.toInt, tp, startingTimestamp =
startingTimestamp)
+ consumer.assign(List(tp).asJava)
+ consumeAndVerifyRecords(consumer = consumer, numRecords =
totalRecords.toInt, startingOffset =0)
+ // seek to out of range position
+ val outOfRangePos = totalRecords + 1
+ consumer.seek(tp, outOfRangePos)
+ // assert that poll resets to the beginning position
+ consumeAndVerifyRecords(consumer = consumer, numRecords = 1,
startingOffset = 0)
+ }
+
+ @Test
+ def testFetchInvalidOffsetResetConfigLatestRecordsBeforeTimeout(): Unit = {
+
+ this.consumerConfig.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
"latest")
+ // long wait time should (theoretically) mean that more records show up
before this expires
+ this.consumerConfig.setProperty(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG,
"3000")
+ val consumer = createConsumer(configOverrides = this.consumerConfig)
+ val totalRecords = 10L
+
+ val producer = createProducer()
+ val startingTimestamp = 0
+ sendRecords(producer, totalRecords.toInt, tp, startingTimestamp =
startingTimestamp)
+ consumer.assign(List(tp).asJava)
+ consumer.seek(tp,0)
+ consumeAndVerifyRecords(consumer = consumer, numRecords =
totalRecords.toInt, startingOffset = 0)
+ // seek to out of range position
+ val outOfRangePos = totalRecords + 7 //arbitrary, much higher offset
+ consumer.seek(tp, outOfRangePos)
+ // assert that poll resets to the ending position
+ assertTrue(consumer.poll(Duration.ofMillis(50)).isEmpty)
+ //some new records show up before fetch max wait time
+ sendRecords(producer, totalRecords.toInt, tp, startingTimestamp =
totalRecords)
+ // ensure that new records start with the offset that was passed to
seek(), as new records have arrived
+ val nextRecord = consumer.poll(Duration.ofMillis(50)).iterator().next()
+ assertEquals(outOfRangePos,nextRecord.offset())
+
+ }
+
+ @Test
+ def testFetchInvalidOffsetResetConfigLatestNoRecordsBeforeTimeout(): Unit = {
+ this.consumerConfig.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
"latest")
+ this.consumerConfig.setProperty(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG,
"0")
+ val consumer = createConsumer(configOverrides = this.consumerConfig)
+ val totalRecords = 10L
+
+ val producer = createProducer()
+ val startingTimestamp = 0
+ sendRecords(producer, totalRecords.toInt, tp, startingTimestamp =
startingTimestamp)
+ consumer.assign(List(tp).asJava)
+ consumer.seek(tp, 0)
+ //consume some, but not all of the records
+ consumeAndVerifyRecords(consumer = consumer, numRecords =
totalRecords.toInt/2, startingOffset = 0)
+ // seek to out of range position
+ val outOfRangePos = totalRecords + 17 //arbitrary, much higher offset
+ consumer.seek(tp, outOfRangePos)
+ // assert that poll resets to the ending position
+ assertTrue(consumer.poll(Duration.ofMillis(50)).isEmpty)
+ sendRecords(producer, totalRecords.toInt, tp, startingTimestamp =
totalRecords)
+ val nextRecord = consumer.poll(Duration.ofMillis(50)).iterator().next()
+ //ensure the seek went to the last known record at the time of the
previous poll
Review Comment:
Please add a space after `//`
##########
core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala:
##########
@@ -694,6 +694,77 @@ class PlaintextConsumerTest extends BaseConsumerTest {
assertEquals(outOfRangePos.toLong, outOfRangePartitions.get(tp))
}
+ @Test
+ def testFetchInvalidOffsetResetConfigEarliest(): Unit = {
+ this.consumerConfig.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
"earliest")
+ val consumer = createConsumer(configOverrides = this.consumerConfig)
+ val totalRecords = 10L
+
+ val producer = createProducer()
+ val startingTimestamp = 0
+ sendRecords(producer, totalRecords.toInt, tp, startingTimestamp =
startingTimestamp)
+ consumer.assign(List(tp).asJava)
+ consumeAndVerifyRecords(consumer = consumer, numRecords =
totalRecords.toInt, startingOffset =0)
+ // seek to out of range position
+ val outOfRangePos = totalRecords + 1
+ consumer.seek(tp, outOfRangePos)
+ // assert that poll resets to the beginning position
+ consumeAndVerifyRecords(consumer = consumer, numRecords = 1,
startingOffset = 0)
+ }
+
+ @Test
+ def testFetchInvalidOffsetResetConfigLatestRecordsBeforeTimeout(): Unit = {
+
+ this.consumerConfig.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
"latest")
+ // long wait time should (theoretically) mean that more records show up
before this expires
+ this.consumerConfig.setProperty(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG,
"3000")
+ val consumer = createConsumer(configOverrides = this.consumerConfig)
+ val totalRecords = 10L
+
+ val producer = createProducer()
+ val startingTimestamp = 0
+ sendRecords(producer, totalRecords.toInt, tp, startingTimestamp =
startingTimestamp)
+ consumer.assign(List(tp).asJava)
+ consumer.seek(tp,0)
+ consumeAndVerifyRecords(consumer = consumer, numRecords =
totalRecords.toInt, startingOffset = 0)
+ // seek to out of range position
+ val outOfRangePos = totalRecords + 7 //arbitrary, much higher offset
+ consumer.seek(tp, outOfRangePos)
+ // assert that poll resets to the ending position
+ assertTrue(consumer.poll(Duration.ofMillis(50)).isEmpty)
+ //some new records show up before fetch max wait time
+ sendRecords(producer, totalRecords.toInt, tp, startingTimestamp =
totalRecords)
+ // ensure that new records start with the offset that was passed to
seek(), as new records have arrived
+ val nextRecord = consumer.poll(Duration.ofMillis(50)).iterator().next()
+ assertEquals(outOfRangePos,nextRecord.offset())
+
+ }
+
+ @Test
+ def testFetchInvalidOffsetResetConfigLatestNoRecordsBeforeTimeout(): Unit = {
+ this.consumerConfig.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
"latest")
+ this.consumerConfig.setProperty(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG,
"0")
Review Comment:
Could we add a comment above this line to mention why we need to set
`FETCH_MAX_WAIT_MS_CONFIG` to 0?
##########
core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala:
##########
@@ -694,6 +694,77 @@ class PlaintextConsumerTest extends BaseConsumerTest {
assertEquals(outOfRangePos.toLong, outOfRangePartitions.get(tp))
}
+ @Test
+ def testFetchInvalidOffsetResetConfigEarliest(): Unit = {
+ this.consumerConfig.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
"earliest")
+ val consumer = createConsumer(configOverrides = this.consumerConfig)
+ val totalRecords = 10L
+
+ val producer = createProducer()
+ val startingTimestamp = 0
+ sendRecords(producer, totalRecords.toInt, tp, startingTimestamp =
startingTimestamp)
+ consumer.assign(List(tp).asJava)
+ consumeAndVerifyRecords(consumer = consumer, numRecords =
totalRecords.toInt, startingOffset =0)
+ // seek to out of range position
+ val outOfRangePos = totalRecords + 1
+ consumer.seek(tp, outOfRangePos)
+ // assert that poll resets to the beginning position
+ consumeAndVerifyRecords(consumer = consumer, numRecords = 1,
startingOffset = 0)
+ }
+
+ @Test
+ def testFetchInvalidOffsetResetConfigLatestRecordsBeforeTimeout(): Unit = {
+
+ this.consumerConfig.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
"latest")
+ // long wait time should (theoretically) mean that more records show up
before this expires
+ this.consumerConfig.setProperty(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG,
"3000")
+ val consumer = createConsumer(configOverrides = this.consumerConfig)
+ val totalRecords = 10L
+
+ val producer = createProducer()
+ val startingTimestamp = 0
+ sendRecords(producer, totalRecords.toInt, tp, startingTimestamp =
startingTimestamp)
+ consumer.assign(List(tp).asJava)
+ consumer.seek(tp,0)
+ consumeAndVerifyRecords(consumer = consumer, numRecords =
totalRecords.toInt, startingOffset = 0)
+ // seek to out of range position
+ val outOfRangePos = totalRecords + 7 //arbitrary, much higher offset
+ consumer.seek(tp, outOfRangePos)
+ // assert that poll resets to the ending position
+ assertTrue(consumer.poll(Duration.ofMillis(50)).isEmpty)
+ //some new records show up before fetch max wait time
+ sendRecords(producer, totalRecords.toInt, tp, startingTimestamp =
totalRecords)
+ // ensure that new records start with the offset that was passed to
seek(), as new records have arrived
+ val nextRecord = consumer.poll(Duration.ofMillis(50)).iterator().next()
+ assertEquals(outOfRangePos,nextRecord.offset())
+
+ }
+
+ @Test
+ def testFetchInvalidOffsetResetConfigLatestNoRecordsBeforeTimeout(): Unit = {
+ this.consumerConfig.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
"latest")
+ this.consumerConfig.setProperty(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG,
"0")
+ val consumer = createConsumer(configOverrides = this.consumerConfig)
+ val totalRecords = 10L
+
+ val producer = createProducer()
+ val startingTimestamp = 0
+ sendRecords(producer, totalRecords.toInt, tp, startingTimestamp =
startingTimestamp)
+ consumer.assign(List(tp).asJava)
+ consumer.seek(tp, 0)
+ //consume some, but not all of the records
Review Comment:
Please add a space after `//`
##########
core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala:
##########
@@ -694,6 +694,77 @@ class PlaintextConsumerTest extends BaseConsumerTest {
assertEquals(outOfRangePos.toLong, outOfRangePartitions.get(tp))
}
+ @Test
+ def testFetchInvalidOffsetResetConfigEarliest(): Unit = {
Review Comment:
nit: Since it's not an invalid offset, it's out of range offset, could we
rename to `testFetchOutOfRangeOffsetResetConfigEarliest`?
##########
core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala:
##########
@@ -672,7 +672,7 @@ class PlaintextConsumerTest extends BaseConsumerTest {
@Test
def testFetchInvalidOffset(): Unit = {
this.consumerConfig.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
"none")
- val consumer = createConsumer()
+ val consumer = createConsumer(configOverrides = this.consumerConfig)
Review Comment:
Nice fix!
##########
clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java:
##########
@@ -1581,6 +1581,23 @@ public void commitAsync(final Map<TopicPartition,
OffsetAndMetadata> offsets, Of
* is invoked for the same partition more than once, the latest offset
will be used on the next poll(). Note that
* you may lose data if this API is arbitrarily used in the middle of
consumption, to reset the fetch offsets
*
+ * The next Consumer Record which will be retrieved when poll() is invoked
will either have the offset specified or,
+ * by default, with {@link ConsumerConfig AUTO_RESET_CONFIG} set to
"latest"
+ * a higher numbered offset. A higher numbered offset will be returned if
there is no consumer record with the offset
+ * specified but there is one with a higher offset.
+ * seek(0) is equivalent to seek to beginning for a topic with beginning
offset 0.
+ * seekToEnd() is equivalent to seeking to the highest known offset + 1.
+ *
+ * Seeking past the end of the highest known offset means an invalid
offset is reached.
Review Comment:
Seeking to the offset earlier than start offset will also return out of
range exception. So, maybe update to:
`Seeking to the offset smaller than the log start offset or larger than the
log end offset or high watermark means an invalid offset is reached`
##########
clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java:
##########
@@ -1581,6 +1581,23 @@ public void commitAsync(final Map<TopicPartition,
OffsetAndMetadata> offsets, Of
* is invoked for the same partition more than once, the latest offset
will be used on the next poll(). Note that
* you may lose data if this API is arbitrarily used in the middle of
consumption, to reset the fetch offsets
*
+ * The next Consumer Record which will be retrieved when poll() is invoked
will either have the offset specified or,
+ * by default, with {@link ConsumerConfig AUTO_RESET_CONFIG} set to
"latest"
+ * a higher numbered offset. A higher numbered offset will be returned if
there is no consumer record with the offset
+ * specified but there is one with a higher offset.
Review Comment:
It's hard to read for these 4 lines. I don't know why we should mention
"latest" reset config here.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java:
##########
@@ -1581,6 +1581,23 @@ public void commitAsync(final Map<TopicPartition,
OffsetAndMetadata> offsets, Of
* is invoked for the same partition more than once, the latest offset
will be used on the next poll(). Note that
* you may lose data if this API is arbitrarily used in the middle of
consumption, to reset the fetch offsets
*
+ * The next Consumer Record which will be retrieved when poll() is invoked
will either have the offset specified or,
+ * by default, with {@link ConsumerConfig AUTO_RESET_CONFIG} set to
"latest"
+ * a higher numbered offset. A higher numbered offset will be returned if
there is no consumer record with the offset
+ * specified but there is one with a higher offset.
+ * seek(0) is equivalent to seek to beginning for a topic with beginning
offset 0.
+ * seekToEnd() is equivalent to seeking to the highest known offset + 1.
+ *
+ * Seeking past the end of the highest known offset means an invalid
offset is reached.
+ * Invalid offset behaviour is controlled by
+ * the {@link ConsumerConfig AUTO_RESET_CONFIG} property. If this is set
to "earliest", the next poll() will seek to the beginning
+ * before returning a record. If it is set to "latest", it will seek to
the last known record (similar to seekToEnd()).
+ * Note that, in the "latest" if any new records show up before {@link
ConsumerConfig FETCH_MAX_WAIT_MS_CONFIG}, then
+ * the next poll() may succeed in seeking to the requested offset, or
otherwise it will seek to the new latest known offset.
Review Comment:
I'm thinking we don't need to mention it here. First, it's not just `latest`
case, `earlier` will also has this behavior. Second, it's too detail to users
to know about it, and make users more confused about it, like so, how should we
do about it? Third, so, back to the point you tried to mentioned, I think
you're trying to explain the case:
1. current log end offset is 10
2. seek to 17
3. poll -> got nothing
4. produce 10 more records
5. poll -> got records offset starting from 17
Is that right? I think that's expected from user's point of view. User tried
to seek to offset 17 and returning records starting from 17, everything works
fine, isn't it?
I think if you really want to talk about it, maybe you can say: the seek
offset won't change to the in-flight fetch request, it will take effect in next
fetch request.
WDYT?
IMO, we don't need to mention it.
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
