了解Kafka消息字节大小

Question

如何获得Kafka单张唱片的大小？

有一些关于我为什么需要这个的说明。

这似乎不是在ConsumerRecord或RecordMetadata类上公开的serializedValueSize。我并不真正理解这个属性的价值，因为它与对消费者有用的消息的大小不匹配。serializedValueSize用于什么？

最大Poll.Records

在对poll（）的单个调用中返回的最大记录数。

这并不存在，但却是我想要的行为：

在消费者大小上，我将“max.partition.fetch.bytes”设置为291字节。这使得消费者只能返回1条消息。将该值设置为292有时会使使用者返回2条消息。所以我已经计算出消息大小为292的一半；一条消息的大小为146字节。

上面的项目符号要求对Kafka配置进行更改，并涉及手动查看/greping一些服务器日志。如果Kafka Java API提供这个值就太好了。

在生产者方面，Kafka提供了一种获取RecordMetadata.SerializedValueSize方法中记录的序列化大小的方法。这个值是76字节，与上面测试中给出的146字节有很大不同。

java prettyprint-override">System.out.println(myRecordMetadata.serializedValueSize());
// 76

# producer
batch.size=1

# consumer

# Expected this to work:
# 76 * 2 = 152
max.partition.fetch.bytes=152

# Actually works:
# 292 = ??? magic ???
max.partition.fetch.bytes=292

我希望将max.partition.fetch.bytes设置为serializedValueSize给出的字节数的倍数可以使Kafka使用者从轮询接收到最多该数量的记录。相反，max.partition.fetch.bytes值需要高得多才能实现这一点。

Answer 1

我不太熟悉serializedvalueSize方法，但根据文档，这只是存储在该消息中的值的大小。这将小于总消息大小（即使使用null键），因为消息还包含不属于值的元数据（如时间戳）。

至于您的问题：与其通过处理消息大小和限制使用者的吞吐量来直接控制轮询，为什么不只是缓冲传入的消息，直到足够的消息可用或所需的超时（您提到了fetch.max.wait.ms，但您可以手动指定一个超时）已经过去？

public static <K, V> List<ConsumerRecord<K, V>>
    minPoll(KafkaConsumer<K, V> consumer, Duration timeout, int minRecords) {
  List<ConsumerRecord<K, V>> acc = new ArrayList<>();
  long pollTimeout = Duration.ofMillis(timeout.toMillis()/10);
  long start = System.nanoTime();
  do {
    ConsumerRecords<K, V> records = consumer.poll(pollTimeout);
    for(ConsumerRecord<K, V> record : records)
      acc.add(record);
  } while(acc.size() < minRecords &&
          System.nanoTime() - start < timeout.toNanos());
  return acc;
}

对consumer.poll的调用中的timeout.ToMillis()/10超时是任意的。您应该选择一个足够小的持续时间，这样即使我们等待的时间比指定的超时时间长（这里是：长10%）也没关系。

编辑：请注意，这可能返回一个大于max.poll.records的列表（最大值为max.poll.records+minrecords-1)。如果您还需要强制执行这个严格的上限，可以使用方法外部的另一个缓冲区临时存储多余的记录（这可能会更快，但不允许minPoll和普通的Poll方法混合使用），或者直接丢弃它们，并使用Consumer的Seek方法进行回溯。

因此，问题不在于控制poll-方法返回的消息数量，而在于如何获得单个记录的大小。不幸的是，我认为那不可能不经过很多麻烦。问题是，对此没有真正的（恒定的）答案，甚至一个大概的答案将取决于Kafka版本，或者更确切地说，不同的Kafka协议版本。

首先，我不完全确定max.partition.fetch.bytes到底控制了什么（如：协议开销是否也是它的一部分？）。让我来说明一下我的意思：当使用者发送一个fetch请求时，那么fetch响应由以下字段组成：

1. 节流时间（4字节）
2. 主题响应数组（4字节表示数组长度+数组中数据的大小）。

主题响应依次包括

1. 主题名称（字符串长度+字符串大小为2个字节）
2. 分区响应数组（4字节表示数组长度+数组中数据的大小）。

null

1. 正文大小（1-5字节）
2. 属性（1字节）
3. 时间戳增量（1-10字节）
4. 偏移增量（1-5字节）
5. 密钥字节数组（1-5字节+密钥数据大小）
6. 值字节数组（1-5字节+值数据大小）
7. 标头（1-5字节+标头数据大小）。

源代码在这里。正如您所看到的，您不能简单地将292字节除以2来获得记录大小，因为某些开销是恒定的，并且与记录的数量无关。

更糟糕的是，记录没有恒定的大小，即使它们的键和值（以及标头）有恒定的大小，因为时间戳和偏移量使用可变长度数据类型存储为与批处理时间戳和偏移量的差异。此外，这只是编写本文时最新协议版本的情况。对于旧版本，答案将再次不同，谁知道会发生什么在未来的版本。