Kafka Streams :如何在Streamer Code - High level consumer中获得轮询的批量大小

Properties props = new Properties();
props.put("bootstrap.servers", "kafka-broker1:9092,kafka-broker2:9092");
props.put("group.id", "my-consumer-group");
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
props.put("session.timeout.ms", "30000");

ConsumerConfig config = new ConsumerConfig(props);
ConsumerConnector consumer = Consumer.createJavaConsumerConnector(config);

Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
topicCountMap.put("my-topic", 1); // 设置要消费的主题和消费者线程数

Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap);

List<KafkaStream<byte[], byte[]>> streams = consumerMap.get("my-topic"); // 获取主题的消息流

for (final KafkaStream<byte[], byte[]> stream : streams) {
    ConsumerIterator<byte[], byte[]> it = stream.iterator();
    while (it.hasNext()) {
        MessageAndMetadata<byte[], byte[]> messageAndMetadata = it.next();
        byte[] message = messageAndMetadata.message();
        int batchSize = message.length; // 获取轮询的批量大小
        // 处理消息
    }
}