Add support for KIP-110 ZSTD compression

[Marek Vavruša (vavrusa)]

This was introduced recently in https://cwiki.apache.org/confluence/display/KAFKA/KIP-110%3A+Add+Codec+for+ZStandard+Compression

And merged in apache/kafka#2267

I'm going to need some help figuring out how to test this. I think I added the prerequisites right - Fetch APIv10 and Produce APIv7.

[Magnus Edenhill (edenhill)]

Wow, that was quick, the AK PR was merged two days ago!
This looks great, just needs a couple of minor cosmetic changes.

[Magnus Edenhill (edenhill)]

should be a break; here, right?

[Magnus Edenhill (edenhill)]

I can't find the source now, and I don't really remember why, but it is supposedly better to multiply by 1.75 than 2 when increasing buffers.

https://github.com/edenhill/librdkafka/blob/master/src/rdkafka_lz4.c#L268

[Magnus Edenhill (edenhill)]

Add this: rd_atomic64_add(&rkb->rkb_c.zbuf_grow, 1);

[Marek Vavruša (vavrusa)]

Updated

[Magnus Edenhill (edenhill)]

variable definitions need to go at the top of the scope to be compatible with older non C99 compilers (MSVC).

[Marek Vavruša (vavrusa)]

I updated it but it doesn't compile with c89 (no inline etc.), is this is just for MSVC compatibility?

[Magnus Edenhill (edenhill)]

Yeah, at least that's the most common weird old compiler we come across frequently, and it is stuck somewhere between c89 and c99 :)

[Magnus Edenhill (edenhill)]

variable definititions need to get at the top of the scope.

err shadows the top-level rd_kafka_resp_err_t err, better call this one zerr

[Magnus Edenhill (edenhill)]

Better to error out here instead of crashing since the input data is from an external source.

[Magnus Edenhill (edenhill)]

dito, rename to rdkafka_zstd.h, add copyright, add trailing >

[Magnus Edenhill (edenhill)]

This file is automatically generated but seems to be missing the zstd stuff; make sure you build the source with make at least once to let auto-generation kick in.

[Magnus Edenhill (edenhill)]

Since we're not including the zstd source we shouldn't include this license.

[Magnus Edenhill (edenhill)]

dito

[Magnus Edenhill (edenhill)]

This will go in v1.0.0, which is currently the idempotence branch.
It would be great if you could retarget and rebase this on idempotence.

[Marek Vavruša (vavrusa)]

Yeah, we've been using zstd in Kafka for a while at Cloudflare, but it only currently "works" with Sarama, so I can't move without it.

[Magnus Edenhill (edenhill)]

.. so I can't move without it.

Move to what? :)

v1.0.0 is scheduled for release within a month.

[Marek Vavruša (vavrusa)]

Move to what? :)

I wrote the Kafka table engine for ClickHouse using librdkafka, but can't use it with most topics without zstd. The librdkafka is embedded as a submodule there, so I don't have to wait for the release tag before testing it.

[Marek Vavruša (vavrusa)]

Rebased to idempotence branch.

[Marek Vavruša (vavrusa)]

Thanks for reviewing! 🙌

[Donis-]

Marek Vavruša (@vavrusa) did you get it to work in your case?

We upgraded to confluent kafka 5.1.0 / 2.1.0 and set the broker default compression to ZSTD.
Java clients seem to be working fine, but i can't get kafkacat with librdkafka to work:
Consumption fails. Based on logs, zstd was enabled via compilation and is supported by broker:

%7|1546470506.414|APIVERSION|rdkafka#consumer-1| [thrd:sasl_plaintext://xxx/bootstrap]: sasl_plaintext://xxx/bootstrap: Feature ZSTD: Produce (7..7) supported by broker
%7|1546470506.415|APIVERSION|rdkafka#consumer-1| [thrd:sasl_plaintext://xxx/bootstrap]: sasl_plaintext://xxx/bootstrap: Feature ZSTD: Fetch (10..10) supported by broker
%7|1546470506.417|APIVERSION|rdkafka#consumer-1| [thrd:sasl_plaintext://xxx/bootstrap]: sasl_plaintext://xxx/bootstrap: Enabling feature ZSTD
%7|1546470506.417|FEATURE|rdkafka#consumer-1| [thrd:sasl_plaintext://xxx/bootstrap]: sasl_plaintext://xxx/bootstrap: Updated enabled protocol features to MsgVer1,ApiVersion,BrokerBalancedConsumer,ThrottleTime,Sasl,SaslHandshake,BrokerGroupCoordinator,LZ4,OffsetTime,MsgVer2,IdempotentProducer,ZSTD
...
%7|1546470508.106|FETCH|rdkafka#consumer-1| [thrd:sasl_plaintext://xxx/105]: sasl_plaintext://xxx/105: Fetch topic XXX [55] at offset 0 (v2)
%7|1546470508.108|FETCH|rdkafka#consumer-1| [thrd:sasl_plaintext://xxx/105]: sasl_plaintext://xxx/105: Fetch 1/1/16 toppar(s)
%7|1546470508.108|SEND|rdkafka#consumer-1| [thrd:sasl_plaintext://xxx/105]: sasl_plaintext://xxx/105: Sent FetchRequest (v4, 112 bytes @ 0, CorrId 3)
%7|1546470508.155|RECV|rdkafka#consumer-1| [thrd:sasl_plaintext://xxx/105]: sasl_plaintext://xxx/105: Received FetchResponse (v4, 92 bytes, CorrId 3, rtt 46.73ms)
%7|1546470508.156|FETCH|rdkafka#consumer-1| [thrd:sasl_plaintext://xxx/105]: sasl_plaintext://xxx/105: Topic XXX [55] MessageSet size 0, error "Err-76?", MaxOffset -1, Ver 2/2

76 is the error introduced with KIP-110: UNSUPPORTED_COMPRESSION_TYPE.
https://cwiki.apache.org/confluence/display/KAFKA/KIP-110%3A+Add+Codec+for+ZStandard+Compression

Maybe we need to bump the v4 fetch version somehow as per "Zstd will only be allowed for the bumped fetch API ".? or am i missing something?

version:
tag 1.0.0-RC5 with dynamic zstd linking, win64 build.

[Marek Vavruša (vavrusa)]

Yes, it's been working for some time. Maybe the negotiation isn't working properly for your case? I don't know that much about Kafka itself to help you, sorry!

[Donis-]

thanks for confirming!
just to double check, you're both producing and consuming via librdkafka?
Magnus Edenhill (@edenhill) maybe you have an idea on what could be going wrong in my case?

[Donis-]

seems fetch api v10 support was missing, by adding the the fetch protocol support for v10 i got it to work. i can clean up and submit a PR.