Can you please store this in an intermediate variable, rather than nesting amatch inside all the error handling?

switch to use a method chain instead of match tree

I'm not sure if it's appropriate to close the connection because of one failed request.

In fact, Kafka 0.8 will close the connection immediately when it received the unknownApiVersionsRequest, we just mark the connection was closed for the connection pool.

Oh, I see. Interesting!

Two questions:

1. Why are theseApiVersions being added to these request structs? If I am not mistaken, the API version goes into the requestheader, not the request itself, so I'm not sure it makes sense to have the API version stored inside every request object.
2. Why are they allRc? This adds significant overhead on every access.ApiVersion is just a simple struct, so even copying it is quite cheap.

1. Yes, there is aapi_version field in the request header, butkafka-rust hard code it to 0. So, we have to add a realApiVersion to decide whether the server use the wrong frame checksum (0.8/0.9) or not (0.10 later).
2. All theRc has been removed

Using the new? operator might make this a little easier to read

I usetry! because the old code use it, if you thinkrust 0.13 later will be fine, I would like to use? in future.

Whoa, this is a lot of low-level code. Did you write this all yourself? If not, where did it come from? Do we need to have all this code, or is there a crate we can use?

Sure, I write it.

In fact, I have extracted it to an isolated cratelz4frame. I'm not sure you like to depend on external crate or embedded those code.

lz4-rs have similar LZ4 frame supports fromliblz4, but it doesn't support the wrong frame checksum used in Kafka 0.8.

Yeah, it doesn't feel appropriate to have all this LZ4 code inside a crate for Kafka, and I think it could significantly increase the burden on maintainers and future contributors who may not have the expertise to mess with the code for an LZ4 implementation. If a library doesn't exist that suits the needs ofkafka-rust, then I think it's more appropriate to wait for one that does than to (please excuse the bluntness, because I really do appreciate your efforts!) have a bunch of hand-written, scantily-tested, -documented, and -commented code to handle a compatibility layer.

To be honest, in the absence of a quality crate that fits our needs, I'd rather just use a crate that uses the right frame checksum and notify users that it will only work for new versions of Kafka.

@xitep what are your thoughts?

ATcpStream is not guaranteed to be open, so I'm not sure it's correct to assume it is here.

it just a mark for the connection pool when we found the server closed the connection. If you like to trace the whole lifecycle of connection, I could add a new state machine for it.

Eh, actually on some more thought, it's probably ok if it turns out that the connection was closed. Further down the chain, it would just return an I/O error I think.

According to the Kafka protocol docs, theApiVersionsRequest is only supported by 0.10.0.0 and up. So if you're just hard-codingAPI_VERSION here, which is in turn hard-coded to the Kafka 0.8 API version, then won't this request fail every time?

Kafka use theapi_version for request, it means different request may have differentapi_version on different server version.

For example,Produce API use

v0, v1 (supported in 0.9.0 or later) and v2 (supported in 0.10.0 or later)

AndOffset API use

v1 (supported in 0.10.1.0 and later)

So, hard code it will be fine for 0.10 or later, and it always failed on Kafka 0.8

Oh, I see; I misunderstood how the API versioning worked. I just realized that every request type has its own individual version, which starts at 0, which will work out here by happenstance. This kind of makes me think that we should refactor the request code in general.

@xitep I think this presents an interesting problem. I was thinking we should make a ticket to update all of our request/response structs with the additional fields from new API versions, but if we do that, how do we deserialize responses from older APIs that don't have those fields? We could useOptions, but that's a very clunky option, I think. We could have separate structs likeProduceResponseV0,ProduceResponseV1, etc., but that's also kind of clunky. Man, maintaining compatibility across all the versions is such a pain 😰