Scanteianu commented on code in PR #12753:
URL: https://github.com/apache/kafka/pull/12753#discussion_r1009200035
##########
core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala:
##########
@@ -694,6 +694,73 @@ class PlaintextConsumerTest extends BaseConsumerTest {
assertEquals(outOfRangePos.toLong, outOfRangePartitions.get(tp))
}
+ @Test
+ def testFetchInvalidOffsetResetConfigEarliest(): Unit = {
+ val consumer = createConsumer()
+ this.consumerConfig.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
"earliest")
+ 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 testFetchInvalidOffsetResetConfigLatest(): Unit = {
+ val consumer = createConsumer()
+ this.consumerConfig.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
"latest")
Review Comment:
so i fixed this, now it seems like the behaviour is much more predictable,
but it still feels like the first `poll` after `seek` to invalid offset should
reset to the current latest unconsumed offset and/or the end of the queue (ieL
offset 10). Nonetheless, this does seem like it might be a conscious design
decision, and therefore all the more important to document.
--
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
