[KAFKA-15652] Add tests to verify OffsetFetcherUtils.getOffsetResetTimestamp()

• Type: Test
• Status: Open
• Priority: Minor
• Resolution: Unresolved
• Fix Version/s: 3.8.0
• Component/s: clients, consumer, unit tests
• Labels: consumer-threading-refactor position unit-tests
• Assignee: Philip Nee
• Reporter: Kirk True
• Created: 19/Oct/23 22:02
• Updated: 01/Feb/24 04:58

Description

In the updateFetchPositions() method implementation, both KafkaConsumer and PrototypeAsyncConsumer reset positions asynchronously. junrao stated the following in a recent PR review:

There is a subtle difference between transitioning to reset from initializing and transitioning to reset from OffsetOutOfRangeException during fetch. In the latter, the application thread will call FetchCollector.handleInitializeErrors(). If there is no default offset reset policy, an OffsetOutOfRangeException will be thrown to the application thread during poll, which is what we want.

However, for the former, if there is no default offset reset policy, we simply ignore that partition through OffsetFetcherUtils.getOffsetResetTimestamp. It seems in that case, the partition will be forever in the reset state and the application thread won't get the OffsetOutOfRangeException.

I intentionally changed the code so that no exceptions were thrown in OffsetFetcherUtils.getOffsetResetTimestamp() and would simply return an empty map. When I ran the unit tests and integration tests, there were no failures, strongly suggesting that there is no coverage of this particular edge case.