Kafka: max.in.flight above 1 can reorder on retry, unless idempotent

Batch 2 can succeed while batch 1 is being retried - unless idempotence tracks sequence numbers.

Code

// Safe combination:
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);

// Without idempotence, only this is safe:
props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 1);

Output

Non-idempotent with 5 in flight:
batch1 fails, batch2 succeeds, batch1 retried and appended AFTER batch2.
The partition is now out of order, permanently.