KAFKA-10842; Use InterBrokerSendThread for raft's outbound network channel

[hachikuji]

This patch contains the following improvements:

• Separate inbound/outbound request flows so that we can open the door for concurrent inbound request handling
• Rewrite KafkaNetworkChannel to use InterBrokerSendThread which fixes a number of bugs/shortcomings
• Get rid of a lot of boilerplate conversions in KafkaNetworkChannel
• Improve validation of inbound responses in KafkaRaftClient by checking correlationId. This fixes a bug which could cause an out of order Fetch to be applied incorrectly.

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[ijuma]

It seems like there are many classes where direct access to data is not actually needed. How do you feel about having a public method in RequestUtils that exposes data for the raft layer instead? Something like:

public static ApiMessage requestData(AbstractRequest req)
public static ApiMessage responseData(AbstractResponse resp)

Or is it not worth it?

[hachikuji]

Yeah, not sure it's worth it, though I don't feel strongly. I think ultimately we're going to start relying more on the generated classes to avoid unnecessary conversions. We're now entering "phase 2" of the request overhaul which means we can start figuring out how to remove the AbstractRequest/AbstractResponse layer. I think it will take more smarts in the generated classes to make a dent here, but if we are agreed on the goal (?), then I do not think preserving the encapsulation here is worthwhile.

[ijuma]

I don't quite understand how we would handle versioning well if we only have data classes. Do you have thoughts on that?

[hachikuji]

Support for optional fields would go a long way I think. I am not sure it will be possible to remove all intermediate representations, but perhaps they can be the exception and not the rule. Some version checks in KafkaApis are probably inevitable.

[ijuma]

I can see this:

1. AbstractRequest/AbstractResponse methods become part of the ApiMessage hierarchy.
2. FooRequest/FooResponse extends FooDataRequest/FooDataResponse (like Colin suggested before)

But I don't think you want to eliminate FooRequest/FooResponse in the example above. You don't need to perform conversions for the inner classes, but it's a place where you can normalize the representation. We do that for many of the existing request/response classes.

[abbccdda]

Just to reply #9732 (comment) here, generateRequests() is part of the InterBrokerSender thread which I didn't write up. In the meantime, I think changing it to one element at a time makes sense for AlterIsr and forwarding cases for now, if we believe that would make the interface easier to be used, let's refactor it.

[hachikuji]

@abbccdda To clarify, what we are asking is why the implementation of generateRequests in BrokerToControllerChannelManager builds a queue for a single item.

[mumrah]

👍

[mumrah]

Is it worth adding a size or isEmpty to UnsentRequests?

[mumrah]

nit: (related to style question elsewhere) if we want to change the style of these class definitions, can we do it as a separate PR? I always find it difficult when style changes are conflated with logical changes

[hachikuji]

That's fair. I actually held myself back. I tried to only touch the cases that I was modifying anyway, but let me know if there are others. This one was especially obnoxious because of the long parameter list to InterBrokerSendThread.

[mumrah]

👍

[mumrah]

I guess we're not sharing these channel managers between anything. In that case, moving them into the classes that need them seems fine.

[mumrah]

Any reason to use Iterator here instead of queue methods (i.e., peek and remove). Is it to ensure a consistent view of the queue while we're going through it?

[hachikuji]

Yeah, I just thought it was a little simpler.