Queue as different kafka produce request if producerId or producerEpoch varies

[akrambek]

No description provided.

[jfallows]

Let's move parameters producerId, producerEpoch and sequence between parameters timestamp and ackMode for consistent ordering with

        struct KafkaProduceDataEx
        {
            int32 deferred = 0;
            int64 timestamp = 0;
            int64 producerId = -1;
            int16 producerEpoch = -1;
            int32 sequence = -1;
            uint32 crc32c = 0;
            KafkaAckMode ackMode = IN_SYNC_REPLICAS;
            KafkaKey key;
            KafkaHeader[] headers;
        }

[jfallows]

Note: the if condition above is checking latestRecordBatch.producerId() == producerId then assigning producerId = latestRecordBatch.producerId() (which is a no-op if they are already same).

[jfallows]

Suggest we adjust the logic to simplify.

First let's conditionally check if a new record batch is needed (if no current batch, producer differs, etc), and create an empty batch (header) if needed, with record count 0 and record batch length for 0 records.

Then, let's dynamically extend the current record batch to increase the record count by 1, the record batch length by new record bytes, plus the record bytes too. This logic should work whether the record batch was just created (empty, no records yet) or already exists with previously encoded records in the batch.

Modifying the record batch record count and length should be feasible with direct update (like crc) instead of rewriting the record batch (header) entirely.