KAFKA-17686: AsyncKafkaConsumer.offsetsForTimes() fails with NullPointerException

[kirktrue]

The code to convert the Map was initially expressed using the Streams API with collect(Collectors.toMap()). Unfortunately, the implementation of Collector returned by Collectors.toMap() does not support null entries. The KafkaConsumer.offsetsForTimes() API explicitly states that nulls can be present in the returned Map, hence this change.

Committer Checklist (excluded from commit message)

• [ ] Verify design and implementation
• [ ] Verify test coverage and CI build status
• [ ] Verify documentation (including upgrade notes)

[chia7712]

KAFKA-14560 will remove the old client protocol API, so do kafka-client 4.0 need to handle null timestamp?

[kirktrue]

KAFKA-14560 will remove the old client protocol API, so do kafka-client 4.0 need to handle null timestamp?

I want to make very sure I understand the KIP:

• Support for ListOffsets v0 is being removed in 4.0
• Timestamp (and Offset) are both required in v1 and later
• The 4.0.0 client should never get to the point where it receives a value of -1 for either the timestamp or offset

Is the above correct?

[kirktrue]

Separate from this PR, it sounds like the client should remove code that supports those outdated RPC versions.

For example, OffsetAndTimestampInternal appears to exist simply “to allow negative timestamps and offset” values. ListOffsetData is another internal class that is used to compensate for missing values in v0 ListOffset RPC responses. Those classes (and surrounding code) should be removed within the AK 4.0.0 timeframe, right?

[chia7712]

Is the above correct?

Great! Glad we're on the same page. :smile:

Those classes (and surrounding code) should be removed within the AK 4.0.0 timeframe, right?

yes, consumer should throw unsupported version if the target broker supports only 0

[kirktrue]

@chia7712 / @lianetm—tests are passing and comments have been addressed. Can you provide another pass of reviews? Thanks!