Add support for KIP-110 ZSTD compression #2053

vavrusa · 2018-10-12T23:31:44Z

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.

edenhill

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

src/rdkafka_msgset_writer.c

edenhill · 2018-10-13T18:29:55Z

src/rdzstd.c

+ case ZSTD_CONTENTSIZE_UNKNOWN:
+ /* Decompressed size cannot be determined, make a guess */
+ out_bufsize = inlen * 2;
+ case ZSTD_CONTENTSIZE_ERROR:


should be a break; here, right?

edenhill · 2018-10-13T18:36:52Z

src/rdzstd.c

+ /* Check if the destination size is too small */
+ ZSTD_ErrorCode err = ZSTD_getErrorCode(ret);
+ if (err == ZSTD_error_dstSize_tooSmall) {
+ out_bufsize *= 2;


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:/edenhill/librdkafka/blob/master/src/rdkafka_lz4.c#L268

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

edenhill · 2018-10-13T18:37:31Z

src/rdzstd.c

+ ZSTD_ErrorCode err = ZSTD_getErrorCode(ret);
+ if (err == ZSTD_error_dstSize_tooSmall) {
+ out_bufsize *= 2;
+ char *newp = rd_realloc(decompressed, out_bufsize);


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

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

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

edenhill · 2018-10-13T18:39:30Z

src/rdzstd.c

+ }
+
+ /* Check if the destination size is too small */
+ ZSTD_ErrorCode err = ZSTD_getErrorCode(ret);


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

edenhill · 2018-10-13T18:44:50Z

src/rdzstd.c

+ }
+ }
+
+ rd_assert(rd_slice_remains(slice) == 0);


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

edenhill · 2018-10-13T18:45:19Z

src/rdzstd.h

@@ -0,0 +1,52 @@
+/*
+ * Copyright (c) 2018, Marek Vavrusa <[email protected]


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

edenhill · 2018-10-13T18:48:01Z

CONFIGURATION.md

@@ -140,3 +140,4 @@ offset.store.method | C | file, broker | broker
 consume.callback.max.messages | C | 0 .. 1000000 | 0 | Maximum number of messages to dispatch in one `rd_kafka_consume_callback*()` call (0 = unlimited) <br>*Type: integer*

 ### C/P legend: C = Consumer, P = Producer, * = both
+


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.

edenhill · 2018-10-13T18:48:28Z

LICENSE.zstd

@@ -0,0 +1,30 @@
+BSD License


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

edenhill · 2018-10-13T18:48:38Z

LICENSES.txt

@@ -342,3 +342,37 @@ For the files wingetopt.c wingetopt.h downloaded from https:/alex85k
 */


+LICENSE.zstd


edenhill · 2018-10-13T18:51:10Z

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.

vavrusa · 2018-10-13T22:53:49Z

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.

edenhill · 2018-10-13T22:56:02Z

.. so I can't move without it.

Move to what? :)

v1.0.0 is scheduled for release within a month.

vavrusa · 2018-10-13T23:01:52Z

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.

vavrusa · 2018-10-13T23:02:07Z

Rebased to idempotence branch.

vavrusa · 2018-10-13T23:02:26Z

Thanks for reviewing! 🙌

Donis- · 2019-01-02T23:19:41Z

@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:

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.

vavrusa · 2019-01-02T23:45:30Z

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- · 2019-01-03T07:45:31Z

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

Donis- · 2019-01-04T17:14:24Z

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.

vavrusa force-pushed the master branch 3 times, most recently from 305e898 to 7efbdfe Compare October 13, 2018 01:45

edenhill suggested changes Oct 13, 2018

View reviewed changes

edenhill added the enhancement label Oct 13, 2018

vavrusa force-pushed the master branch from 7efbdfe to a9965a0 Compare October 13, 2018 22:52

vavrusa changed the base branch from master to idempotence October 13, 2018 22:54

Clarified regex support in subscribe() (confluentinc#2026)

5664e12

vavrusa force-pushed the master branch from a9965a0 to 8f08520 Compare October 13, 2018 22:56

vavrusa force-pushed the master branch from 8f08520 to 55c14d8 Compare October 14, 2018 18:12

edenhill added this to the v1.0.0 milestone Oct 14, 2018

vavrusa force-pushed the master branch from 55c14d8 to 63b2bba Compare October 15, 2018 21:54

Add support for KIP-110 ZSTD compression

2b21c18

vavrusa force-pushed the master branch from 63b2bba to 2b21c18 Compare October 17, 2018 05:39

edenhill changed the base branch from idempotence to master October 22, 2018 11:21

edenhill approved these changes Oct 22, 2018

View reviewed changes

edenhill merged commit 5475b60 into confluentinc:master Oct 22, 2018

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add support for KIP-110 ZSTD compression #2053

Add support for KIP-110 ZSTD compression #2053

vavrusa commented Oct 12, 2018

edenhill left a comment

edenhill Oct 13, 2018

edenhill Oct 13, 2018

edenhill Oct 13, 2018

vavrusa Oct 13, 2018

edenhill Oct 13, 2018

vavrusa Oct 13, 2018

edenhill Oct 13, 2018

edenhill Oct 13, 2018

edenhill Oct 13, 2018

edenhill Oct 13, 2018

edenhill Oct 13, 2018

edenhill Oct 13, 2018

edenhill Oct 13, 2018

edenhill commented Oct 13, 2018

vavrusa commented Oct 13, 2018

edenhill commented Oct 13, 2018

vavrusa commented Oct 13, 2018

vavrusa commented Oct 13, 2018

vavrusa commented Oct 13, 2018

Donis- commented Jan 2, 2019 •

edited

Loading

vavrusa commented Jan 2, 2019

Donis- commented Jan 3, 2019

Donis- commented Jan 4, 2019

		@@ -0,0 +1,52 @@
		/*
		* Copyright (c) 2018, Marek Vavrusa <[email protected]

		@@ -140,3 +140,4 @@ offset.store.method \| C \| file, broker \| broker
		consume.callback.max.messages \| C \| 0 .. 1000000 \| 0 \| Maximum number of messages to dispatch in one `rd_kafka_consume_callback()` call (0 = unlimited) <br>Type: integer*

		### C/P legend: C = Consumer, P = Producer, * = both

		@@ -342,3 +342,37 @@ For the files wingetopt.c wingetopt.h downloaded from https:/alex85k
		*/


		LICENSE.zstd

Add support for KIP-110 ZSTD compression #2053

Add support for KIP-110 ZSTD compression #2053

Conversation

vavrusa commented Oct 12, 2018

edenhill left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

edenhill commented Oct 13, 2018

vavrusa commented Oct 13, 2018

edenhill commented Oct 13, 2018

vavrusa commented Oct 13, 2018

vavrusa commented Oct 13, 2018

vavrusa commented Oct 13, 2018

Donis- commented Jan 2, 2019 • edited Loading

vavrusa commented Jan 2, 2019

Donis- commented Jan 3, 2019

Donis- commented Jan 4, 2019

Donis- commented Jan 2, 2019 •

edited

Loading