Spring Kafka 使用者无法消费记录

Question

我们使用 Spring Kafka 批量使用记录。我们有时会遇到一个问题，即应用程序启动并且即使有足够的未读消息也不会消耗任何记录。相反，我们不断看到信息日志说。

[INFO]-[FetchSessionHandler:handleError:440] - [Consumer clientId=consumer-2, groupId=groupId] Error sending fetch request (sessionId=INVALID, epoch=INITIAL) to node 1027: org.apache.kafka.common.errors.DisconnectException. 

人们面临这个问题，每个人都说忽略它，因为它只是一个信息日志。甚至，我们看到一段时间后应用程序开始拾取记录而不执行任何操作。但是，开始消费记录可能需要多长时间是非常不可预测的:(

我们在使用 Spring 云流时没有看到此错误。不确定我们是否错过了 spring-kafka 中的任何配置。

过去有人遇到过这个问题，如果我们遗漏了什么，请告诉我们。我们的主题负载巨大，如果有很多滞后，会发生这种情况吗？

我们正在使用Spring Kafka 的 2.2.2.RELEASESpring boot 2.1.2.RELEASEKafka 0.10.0.1(我们知道它已经很老了，因为不可避免的原因我们不得不使用这个:()

这是我们的代码：

应用程序.yml

li.topics: CUSTOM.TOPIC.JSON
spring:
application:
name: DataPublisher
kafka:
listener:
type: batch
ack-mode: manual_immediate
consumer:
enable-auto-commit: false
max-poll-records: 500
fetch-min-size: 1
fetch-max-wait: 1000
group-id: group-dev-02
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer:CustomResourceDeserialiser
auto-offset-reset: earliest

消费者：

public class CustomKafkaBatchConsumer {

@KafkaListener(topics = "#{'${li.topics}'.split(',')}", id = "${spring.kafka.consumer.group-id}")
public void receiveData(@Payload List<CustomResource> customResources,
Acknowledgment acknowledgment,
@Header(KafkaHeaders.RECEIVED_PARTITION_ID) List<Integer> partitions,
@Header(KafkaHeaders.OFFSET) List<Long> offsets) {
}
}

反序列化器：

public class CustomResourceDeserialiser implements Deserializer<CustomResource> {
@Override
public void configure(Map<String, ?> configs, boolean isKey) {
}
@Override
public CustomResource deserialize(String topic, byte[] data) {
if (data != null) {
try {
ObjectMapper objectMapper = ObjectMapperFactory.getInstance();
return objectMapper.readValue(data, CustomResource.class);
} catch (IOException e) {
log.error("Failed to deserialise with {}",e.getMessage());
}
}
return null;
}
@Override
public void close() {
}
}

Answer 1

这可能是因为这个 Kafka-8052 - FETCH 请求问题的间歇性INVALID_FETCH_SESSION_EPOCH错误。这在 Kafka 2.3.0 中已修复

不幸的是，截至 2019 年 8 月 21 日，Spring 云流尚未升级其依赖项，尚未发布 kafka 客户端的 2.3.0。

您可以尝试将这些作为显式依赖项添加到您的 gradle 中

compile ('org.apache.kafka:kafka-streams:2.3.0')
compile ('org.apache.kafka:kafka-clients:2.3.0')
compile ('org.apache.kafka:connect-json:2.3.0')
compile ('org.apache.kafka:connect-api:2.3.0')

更新

这也可能是由 kafka 代理 - 客户端不兼容引起的。如果您的集群落后于客户端版本，您可能会看到各种奇怪的问题，例如。例如，假设您的 kafka 代理在 1.x.x 上，而您的 kafka-consumer 在 2.x.x 上，这可能会发生

Answer 2

我以前遇到过同样的问题，解决方案是减少当前分区计数或增加消费者数量。就我而言，我们在 100 分区上有 ~60M 数据，并且在单个 pod 运行时遇到了相同的错误。我扩展了 30 个 pod(30 个消费者)，问题就解决了。