Produce requests out of order - unspecificied by the actual protocol - upstream with queue depth of 1

[mmaslankaprv]

Test number 0044 from librdkafka test suite assumes that order of produce requests will be maintained and batches will be inserted into log in produce requests order. In redpanda we process requests asynchronously and they may be scheduled in random order by seastar scheduler. On the one hand this behavior is fine since we return offset in produce response.

Here is an example output from the test failing:

[0041_fetch_max_bytes        /  4.608s] CONSUME: consume_msgs: rdkafkatest_rnd7d1cd6ac2df1c72a_0041_fetch_max_bytes [0]: expected msg #1285 (1285/2000): got msg #1283
[0041_fetch_max_bytes        /  4.608s] CONSUME: consume_msgs: rdkafkatest_rnd7d1cd6ac2df1c72a_0041_fetch_max_bytes [0]: expected msg #1287 (1287/2000): got msg #1288
[0041_fetch_max_bytes        /  4.608s] CONSUME: consume_msgs: rdkafkatest_rnd7d1cd6ac2df1c72a_0041_fetch_max_bytes [0]: expected msg #1288 (1288/2000): got msg #1287
[0041_fetch_max_bytes        /  4.609s] CONSUME: consume_msgs: rdkafkatest_rnd7d1cd6ac2df1c72a_0041_fetch_max_bytes [0]: expected msg #1293 (1293/2000): got msg #1294
[0041_fetch_max_bytes        /  4.609s] CONSUME: consume_msgs: rdkafkatest_rnd7d1cd6ac2df1c72a_0041_fetch_max_bytes [0]: expected msg #1294 (1294/2000): got msg #1296
[0041_fetch_max_bytes        /  4.609s] CONSUME: consume_msgs: rdkafkatest_rnd7d1cd6ac2df1c72a_0041_fetch_max_bytes [0]: expected msg #1295 (1295/2000): got msg #1293
[0041_fetch_max_bytes        /  4.610s] CONSUME: consume_msgs: rdkafkatest_rnd7d1cd6ac2df1c72a_0041_fetch_max_bytes [0]: expected msg #1296 (1296/2000): got msg #1295
[0041_fetch_max_bytes        /  4.611s] CONSUME: consume_msgs: rdkafkatest_rnd7d1cd6ac2df1c72a_0041_fetch_max_bytes [0]: expected msg #1306 (1306/2000): got msg #1307
[0041_fetch_max_bytes        /  4.611s] CONSUME: consume_msgs: rdkafkatest_rnd7d1cd6ac2df1c72a_0041_fetch_max_bytes [0]: expected msg #1307 (1307/2000): got msg #1308
[0041_fetch_max_bytes        /  4.611s] CONSUME: consume_msgs: rdkafkatest_rnd7d1cd6ac2df1c72a_0041_fetch_max_bytes [0]: expected msg #1308 (1308/2000): got msg #1306
[0041_fetch_max_bytes        /  4.612s] CONSUME: consume_msgs: rdkafkatest_rnd7d1cd6ac2df1c72a_0041_fetch_max_bytes [0]: expected msg #1313 (1313/2000): got msg #1314
[0041_fetch_max_bytes        /  4.612s] CONSUME: consume_msgs: rdkafkatest_rnd7d1cd6ac2df1c72a_0041_fetch_max_bytes [0]: expected msg #1314 (1314/2000): got msg #1316
[0041_fetch_max_bytes        /  4.613s] CONSUME: consume_msgs: rdkafkatest_rnd7d1cd6ac2df1c72a_0041_fetch_max_bytes [0]: expected msg #1315 (1315/2000): got msg #1313
[0041_fetch_max_bytes        /  4.613s] CONSUME: consume_msgs: rdkafkatest_rnd7d1cd6ac2df1c72a_0041_fetch_max_bytes [0]: expected msg #1316 (1316/2000): got msg #1318
[0041_fetch_max_bytes        /  4.613s] CONSUME: consume_msgs: rdkafkatest_rnd7d1cd6ac2df1c72a_0041_fetch_max_bytes [0]: expected msg #1317 (1317/2000): got msg #1315
[0041_fetch_max_bytes        /  4.613s] CONSUME: consume_msgs: rdkafkatest_rnd7d1cd6ac2df1c72a_0041_fetch_max_bytes [0]: expected msg #1318 (1318/2000): got msg #1317

During the standup @ April 22nd we talked about providing configuration option to control reordering of Kafka messages processing. Three options were proposed

• leave it as it is now
• process all requests on shard in order
• keep order of casually related requests f.e. all produce requests from the same connection for the same NTP

[mmaslankaprv]

I've tested it with kafka serever queue depth limited to 1. This guarantee that requests are processed in order. This make the test to PASS.

[emaxerrno]

this doesn't make sense to me, the client doesn't move to the next batch unless acks=0, until the next batch was acknowledged for acks=0 or -1

[emaxerrno]

ah i see. good point. i think the core-local dispatcher can pipeline the requests too. wondering if the cluster::paritition should handle that

maybe the dispatcher core will add a sequencer and the batch_replicator in raft will enqueue in order

[dotnwat]

@senior7515 initially the sequencing needs to happen at the kafka layer. for example, we can allow two produce requests to run in parallel if they operate on different partitions, but operations on the same partition should be sequenced. if we enforced this below the kafka layer then we wouldn't be able to control interleaving of things like produce and fetch. those aren't inherently problematic, but all of the kafka handlers have been implicitly written with the assumption that they don't run in parallel with other handlers that might have data dependencies. the problem isn't memory safety or anything like that, it is more about consistency and ordering.

at the very least we should probably add this control in under a feature flag. someone is bound to run up against this as @mmaslankaprv has already seen it in some test cases in librdkafka.

[twmb]

we may return for instance produce requests that were sent in order but with replies whose last offsets are out of order, as a result

This may break clients for producing, and I'd definitely have to check my own client to see if it'd handle this type of logic. Users may rely on records being processed in order per partition (if they are produced successfully), so it would be anomalous to produce in order in a client but actually have the broker handle the requests out of order and for the responses to be out of order. I'm also not sure how out of order processing can work with idempotent / transactional producers, because both of these have sequence numbers embedded in the request, and the server must process these requests in order?

[dotnwat]

@twmb thanks for chiming in. I've asked @rystsov about this in the context of idempotent producers and it seems that the sequence number handling internally should identify and handle out-of-order requests. @rystsov any insight on this for transactions?

@twmb I'm curious what your thoughts are around allowing out of order requests for requests that operate on disjoint sets of partitions. we'd still return the results in order to the client, but internally we might extract some parallelism.

[twmb]

re: sequence number handling, I think that handling an idempotent producer forces an individual partition to be processed in order. Request A may have offsets 1 through 10 for a partition, and request B 11 through 20 for that partition, the broker can't handle request B before request A because whatever is tracking sequence numbers will think that B is out of order, where from the client it is not. The client will receive response A that indicates 1 through 10 were produced successfully, but then response B will say OutOfOrderSequenceNumber. From a client's perspective, this implies data loss has occurred, because the client has to assume that the broker lost data for offsets before 11.

For requests that operate on completely disjoint sets of partitions, I think it should be fine to process them out of order. The only ordering that I know of that's important is per-partition ordering. If req A has partition 1, req B has partition 2, and req C has partition 1, then A and C must be processed in order, but B can be processed at any time (so long as the responses are in order, due to the correlation ID).

I think it's should also be possible to split an incoming request into its constituent partitions and handle these as separate single-partition requests, so long as the response is stitched together properly (i.e., translate a multi-partition produce into individual partitions to allow for the out-of-order disjoint example you brought up).

[dotnwat]

We now process requests in order.

[dotnwat]

#1496