获取Kafka压缩消息大小

Question

我想知道Kafka中信息的压缩大小。

我使用kafka 1.1.0和java kafka connect 1.1.0从我的制作人向主题发送消息。

如果消息对我的制作人来说太大，我会得到一个

消息序列化时为xxx字节，大于使用max.request配置的最大请求大小。大小配置。

设置最大请求。将大小设置为合适的值将导致来自代理的错误消息作为消息。代理配置中的max.bytes也必须相应地进行调整。不幸的是，错误消息没有包括代理收到的消息的大小。我调整了消息。最大字节数。到现在为止，一直都还不错。

如果我在生产者端激活压缩，最大请求。大小仍然必须与未压缩的大小相同，因为在压缩之前，代码会比较未压缩消息的大小（请参阅）https://issues.apache.org/jira/browse/KAFKA-4169)

但通过压缩，我可以减少信息量。代理中的最大字节数。问题是，在任何时候我都无法确定这个压缩消息的大小。有没有办法在发送消息之前在生产商代码中或稍后在日志文件中找到答案？

在我的压缩示例中，消息的默认值为1MB。max.bytes就足够了，所以我不必更改默认配置。但我想知道我的压缩信息是远低于1MB还是仅仅0.99MB。在这种情况下，我可能会增加消息。生产中的最大字节数，以避免出现问题。

提前感谢您的支持。

Answer 1

为了测试snappy压缩消息，您可以执行以下操作。

pip install python-snappy
python -m snappy -c input.json output.snappy

Answer 2

你可以使用压缩库，自己压缩信息，在发送之前检查大小。例如，假设您使用的是lz4压缩，那么您可以使用lz4 java lib，然后使用类似于：

private static LZ4Compressor COMPRESS = LZ4Factory.fastestInstance().highCompressor();

String meMessageString      = "My Message that I am sending to kafka";
byte[] uncompressedBytes    = jsonRequest.getBytes();
long lz4compressedLength    = COMPRESSOR.compress(uncompressedBytes).length;