Align KIP-79; Redpanda shouldn't return negative value when retention expires

[daisukebe]

Version & Environment

Redpanda version: v21.11.12 (also exists in the dev branch, 248e103)

What went wrong?

Current Redpanda doesn't support what KIP-79 has carried, especially

The target timestamp have to be non-negative. Otherwise an IllegalArgumentException will be thrown. If no message has a timestamp that is greater than or equals to the target time, a null will be returned.

Once a record's retention gets expired, offsetsForTimes throws java.lang.IllegalArgumentException: Invalid negative timestamp since Redpanda returns a negative value.

What should have happened instead?

Redpanda should return non-negative value

How to reproduce the issue?

1. Let a record's retention get expired
2. Call offsetsForTimes

Additional information

java.lang.IllegalArgumentException: Invalid negative timestamp
	at org.apache.kafka.clients.consumer.OffsetAndTimestamp.<init>(OffsetAndTimestamp.java:39)
	at org.apache.kafka.clients.consumer.internals.Fetcher.offsetsForTimes(Fetcher.java:522)
	at org.apache.kafka.clients.consumer.KafkaConsumer.offsetsForTimes(KafkaConsumer.java:2116)
	at org.apache.kafka.clients.consumer.KafkaConsumer.offsetsForTimes(KafkaConsumer.java:2080)
	at redpanda.Offset.main(Offset.java:42)

[michaelandrepearce]

tbh clear its not just for retention expired, if you set a search for a time where the partition doesnt have data after or at the timestamp it also incorrectly returns -1 and shouldn't it should return as null

[michaelandrepearce]

From KIP-79 "If no message has a timestamp that is greater than or equals to the target time, a null will be returned"

[LenaAn]

Here's my findings so far:

Kafka Protocol:
the time in ListOffsetRequest means different things in v0 and starting v1

In ListOffsetRequest/ListOffsetResponse v0, we return a list of offsets which is smaller than or equals to the target time.

Starting v1 (from KIP-79):

Look up the offsets for the given partitions by timestamp. The returned offset for each partition is the earliest offset whose timestamp is greater than or equals to the given timestamp in the corresponding partition. If no message has a timestamp that is greater than or equals to the target time, a null will be returned

Current behavior of Redpanda:

1. We don't support v0 (but we claim to), we always try to return the earliest offset whose timestamp is greater than or equals to the given timestamp.
2. If there's no such offset, we return last_stable_offset with timestamp(-1)

[LenaAn]

/backport v21.11.x