Fix handling of invalid base offsets

[rodaine]

When a successfully produced batch is returned with an invalid base offset (-1) but no error, the records' internal offsets are being miscomputed from this value. An invalid offset is returned with a non-errored batch primarily in situations where a duplicate idempotent produce is detected, but the actual base offset cannot be provided. Apache Kafka's producer client handles this by setting all record offsets to -1 if the batch's base offset is -1.

This patch makes this correction in the kgo producer callback logic, as well as in the recordToRecord function (though an invalid offset is probably impossible on the fetch side).

[twmb]

The more you know. I'll push a patch release with this soon.

[richardartoul]

What scenario would this happen in? I can't find anywhere in the Kafka codebase where the DuplicateSequence exception is actually thrown, and in the correct case where the client only has 5 outstanding idempotent batches, the Kafka broker will return the correct offsets/timestamp for duplicate batches

[twmb]

(I'm also interested if there's an answer for the above question)

[rodaine]

Honestly I can't say definitively, as this was triggered through some long-running fuzz tests against a very unhealthy cluster. Regardless, this patch improves kgo's behavior to never return erroneous offsets on records under any circumstances, which I feel is generally good.

[rodaine]

(To be clear, I think neither Apache Kafka nor RedPanda have ever returned the duplicate sequence number error.)