Processing requests in order in Kafka layer

[mmaslankaprv]

Cover letter

Changed the way how we process requests in Redpanda Kafka protocol implementation. Previously we processed requests in background, this allowed us to process multiple requests at a time but it may cause issues since requests were not guaranteed to be processed in the same order as they were received. Implemented handling Kafka requests in foreground, this way we normally process one request at a time per connection in exactly the same order as the requests were received.

Produce and commit offsets requests are treated differently as they leverage two phase processing. We process request dispatch in foreground, after this first phase request processing order is guaranteed. Then the second phase is processed in background. This approach allow us to process multiple produce & offset commit requests at a time without compromising ordering.

Fixes: N/A

Release notes

• Full compatibility with Kafka with regards to requests ordering.

[dotnwat]

this is awesome and I think its really close to going in. I think the only real important thing in my feedback in (1) the setting of the exception that crosses cores (see produce.cc feedback) I'm not sure its an issue but its worth looking at closely, and (2) it seems like all of the cases where the cross-core promise is set that that cross-core traffic could be done in the background (without appropriate tracking via gates). but (2) is probably a future optimization. (3) can we add a comment to group_commit / produce about how the cross core promise works (potentially removing the foreign pointer) and stating what the rules are for interacting with that pointer/promise?

[dotnwat]

I'm not sure a foreign ptr here is necessary since you explicitly handle the cross-core stuff yourself. maybe consider using a normal unique ptr? in either case, I do think that there should be a short comment here about how the cross-core promise signaling works

[mmaslankaprv]

I did previously used std::unique_ptr the problem is that lambda is handled on foreign core, deletion of lambda captures is done on the remote core, it causes segmentation faults.

[dotnwat]

it sounds like the segmentation fault is a race condition? normal memory can be freed on a remote core without a foreign pointer which is an optimization.

[mmaslankaprv]

it is indeed a race condition, using std::unique_ptr across shards isn't safe, logic inside the std::unique_ptr is triggered twice and it may lead to use after free situations.

[dotnwat]

Can you explain? AFAICT foreign pointer should be an optimization, unnecessary for correctness. Thus, it sounds like there is a bug.

[dotnwat]

Summary from out-of-band conversation:

hypothesis: the unique pointer is actually destroyed on the destination core (not the source core where the promise is created/set). in this case the promise destructor detects the issue and has a problem. the foreign pointer masks that, doing double duty as an optimization and correctness. I think that in this case it is less of an optimization since we are adding additional round-trips.

in submit_to(source_core, ...) when we set the promise value, it seems like we could at that moment also take care of reseting the unique_ptr and destroying the promise on the correct core (basically the same thing that foreign pointer would be doing).

[dotnwat]

I tried this and I didn't have any issues, but I'm not really sure how to reproduce the issue you were seeing when the foreign pointer wasn't being used.

also, after res.dispatched completes, can we also background the promise sets (not just in the error cases?)

diff --git a/src/v/kafka/server/handlers/produce.cc b/src/v/kafka/server/handlers/produce.cc
index 8e0495caa..34aa9879a 100644
--- a/src/v/kafka/server/handlers/produce.cc
+++ b/src/v/kafka/server/handlers/produce.cc
@@ -237,7 +237,7 @@ static partition_produce_stages produce_topic_partition(
     auto reader = reader_from_lcore_batch(std::move(batch));
     auto start = std::chrono::steady_clock::now();
 
-    auto dispatch = ss::make_foreign<std::unique_ptr<ss::promise<>>>(
+    auto dispatch = std::unique_ptr<ss::promise<>>(
       std::make_unique<ss::promise<>>());
     auto dispatch_f = dispatch->get_future();
     auto f
@@ -259,6 +259,7 @@ static partition_produce_stages produce_topic_partition(
                     (void)ss::smp::submit_to(
                       source_shard, [dispatch = std::move(dispatch)]() mutable {
                           dispatch->set_value();
+                          dispatch.reset();
                       });
                     return ss::make_ready_future<produce_response::partition>(
                       produce_response::partition{
@@ -270,6 +271,7 @@ static partition_produce_stages produce_topic_partition(
                     (void)ss::smp::submit_to(
                       source_shard, [dispatch = std::move(dispatch)]() mutable {
                           dispatch->set_value();
+                          dispatch.reset();
                       });
                     return ss::make_ready_future<produce_response::partition>(
                       produce_response::partition{
@@ -288,18 +290,22 @@ static partition_produce_stages produce_topic_partition(
                   .then_wrapped([source_shard, dispatch = std::move(dispatch)](
                                   ss::future<> f) mutable {
                       if (f.failed()) {
-                          return ss::smp::submit_to(
+                          (void)ss::smp::submit_to(
                             source_shard,
                             [dispatch = std::move(dispatch),
                              e = f.get_exception()]() mutable {
                                 dispatch->set_exception(e);
+                                dispatch.reset();
                             });
+                          return ss::now();
                       }
-                      return ss::smp::submit_to(
+                      (void)ss::smp::submit_to(
                         source_shard,
                         [dispatch = std::move(dispatch)]() mutable {
                             dispatch->set_value();
+                            dispatch.reset();
                         });
+                      return ss::now();
                   })
                   .then([f = std::move(stages.produced)]() mutable {
                       return std::move(f);

[mmaslankaprv]

I was seeing an error after this code was deployed to the cluster and executed for a while. I am going to check this

[dotnwat]

I guess it is reasonable to use foreign pointer as an RAII tool for cases like moving around the promise which will panic if destroyed on the other core. but it was good to get to the bottom of the reasoning behind things being fragile.

looking again, I think the right solution is to use the foreign pointer (for the RAII protection against exception paths), but to do the explicit reset on the source core to avoid the extra core round-trip. that extra round trip should be avoided in foreign pointer destructor if it finds that the pointer had already been reset.