Spring kafka不支持大型消息使用者

Question

我正在使用spring Kafka来使用LinkedIn支持大型消息的Kafka客户端生成的消息

假设此Kafka客户端总是将auto_offset_reset_config重写为none（如其构造函数所示）。

private LiKafkaConsumerImpl(LiKafkaConsumerConfig configs,
    Deserializer<K> keyDeserializer,
    Deserializer<V> valueDeserializer,
    Deserializer<LargeMessageSegment> largeMessageSegmentDeserializer,
    Auditor<K, V> consumerAuditor) {
        _kafkaConsumer = new KafkaConsumer<>(configs.configForVanillaConsumer(),
        byteArrayDeserializer,
        byteArrayDeserializer);
    }
Map<String, Object> configForVanillaConsumer() {
    Map<String, Object> newConfigs = new HashMap<>();
    newConfigs.putAll(this.originals());
    newConfigs.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
    newConfigs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "none");
    return newConfigs;
}

因此，一旦我开始使用批处理提交并将enable_auto_commit_config设置为false，它将抛出以下错误：

发生此问题是因为这是此使用者组第一次使用来自此主题的消息，因此它尝试使用偏移量重置策略。

虽然我将其设置为“最早”，但LinkedIn kafka客户端将其覆盖为“无

在本例中，我还尝试重写ConsumerRebalanceListener，以手动搜索到开头，但实际上并没有到这一步。

Answer 1

有趣的；请在GitHub中打开一个问题。

如果策略是none,我们应该捕获该异常。

同时，您可能只需要使用一次常规客户端就可以解决这一问题，为组设置初始偏移量（您不必实际接收任何消息，只需分配分区并设置组的初始位置）。