Apache Kafka 的 Spring：如何查找分区的 seekToEnd？

Question

我正在使用 Spring Boot 2.0.2.RELEASE 和 Spring for Apache Kafka（有效的 pom 显示 2.1.6.RELEASE 用于 spring-kafka 版本）。

我已经从使用普通的ByteArrayDeserializer 变成了使用 Confluent 的反序列化器

properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, io.confluent.kafka.serializers.KafkaAvroDeserializer.class);

因此，我不必获取字节然后将它们反序列化到有效负载等中。但它的副作用是一些旧消息 - 我无法再阅读，因为它们的架构在融合注册表。

所以当我启动应用程序时，我不断收到这条消息

    2018-08-17 17:58:51.360 ERROR 18004 --- [ntainer#0-0-C-1] o.s.k.listener.BatchLoggingErrorHandler  : Error while processing:

org.apache.kafka.common.errors.SerializationException: Error deserializing key/value for partition ri00-q-log-et-final-0 at offset 36833. If needed, please seek past the record to continue consumption.
Caused by: org.apache.kafka.common.errors.SerializationException: Error deserializing Avro message for id -1
Caused by: org.apache.kafka.common.errors.SerializationException: Unknown magic byte!

所以我决定我必须从主题的结尾开始收听，我检查了文档https://docs.spring.io/spring-kafka/reference/htmlsingle/#seek 建议实现 ConsumerSeekAware 及其子接口 ConsumerSeekAware.ConsumerSeekCallback

我更改了包含@KafkaListener 方法的@service 类 实现文档中提到的接口

@Service
 public class MyAvroListener implements 
 ConsumerSeekAware.ConsumerSeekCallback,ConsumerSeekAware {

它有 @kafkalistener 注释的方法，我尝试过寻找分区的结尾

@KafkaListener(topics = "${topic}", containerFactory = "myAvroListenerFactory")
    public void listen(final Acknowledgment ack, final List<ConsumerRecord<String, EclLogging>> messages) throws Exception {
    this.seekCallBack.get().seekToEnd(topic,0);
    try {
        for (ConsumerRecord<String, EclLogging> kafkaRecord : messages) {

我也尝试过寻找特定的偏移量（因为我一直卡在 36833 偏移量消息）

@KafkaListener(topics = "${topic}", containerFactory = "myAvroListenerFactory")
        public void listen(final Acknowledgment ack, final List<ConsumerRecord<String, EclLogging>> messages) throws Exception {
        this.seekCallBack.get().seek(topic,0,36900);
        try {
            for (ConsumerRecord<String, EclLogging> kafkaRecord : messages) {

我已经实现了上述接口的方法

private final ThreadLocal<ConsumerSeekCallback> seekCallBack = new ThreadLocal<>();

@Override
public void registerSeekCallback(ConsumerSeekCallback consumerSeekCallback) {
    this.seekCallBack.set(consumerSeekCallback);
}

@Override
public void onPartitionsAssigned(Map<TopicPartition, Long> map, ConsumerSeekCallback consumerSeekCallback) {

}

@Override
public void onIdleContainer(Map<TopicPartition, Long> map, ConsumerSeekCallback consumerSeekCallback) {

}

@Override
public void seek(String s, int i, long l) {

}

@Override
public void seekToBeginning(String s, int i) {

}

@Override
public void seekToEnd(String topic, int partition) {
    System.out.println("seekToEnd is hit for topic s = " + topic + " and partition i=" + partition);
}

当应用程序启动时， registerSeekCallBack 方法确实被命中， 但是 seekToEnd 或 seek 方法没有被击中。

因此我不断收到此消息

    2018-08-17 17:58:51.360 ERROR 18004 --- [ntainer#0-0-C-1] o.s.k.listener.BatchLoggingErrorHandler  : Error while processing:

org.apache.kafka.common.errors.SerializationException: Error deserializing key/value for partition ri00-q-log-et-final-0 at offset 36833. If needed, please seek past the record to continue consumption.
Caused by: org.apache.kafka.common.errors.SerializationException: Error deserializing Avro message for id -1
Caused by: org.apache.kafka.common.errors.SerializationException: Unknown magic byte!

我正在使用这里的 sn-p Spring Kafka Template implementaion example for seek offset, acknowledgement

正如这里提到的What determines Kafka consumer offset?，我不能使用 auto.offset.reset 属性从主题末尾开始消费（除非我使用不同的 consumerGroupId - 在我的情况下这是不可能的）。我想知道是否可以使用现有的消费者组来解决这个问题。

Answer 1

您执行搜索太晚了 - 在获取记录的poll() 之后；您需要在

中执行搜索

@Override
public void onPartitionsAssigned(Map<TopicPartition, Long> map, ConsumerSeekCallback consumerSeekCallback) {

}

在那里打电话给consumerSeekCallback.seekToEnd(...)。搜索将在poll() 之前发生以获取记录。

您还可以使用kafka-consumer-groups 命令行工具为组/主题/分区设置任意偏移量。

当前启动版本是2.0.4，kafka 2.1.8。

另外，你不应该实现传递给你的回调。

文档似乎很清楚...

当使用组管理时，当分配改变时调用第二种方法。您可以使用此方法，例如，通过调用回调设置分区的初始偏移量；您必须使用回调参数，而不是传递给 registerSeekCallback 的参数。

...如果不是，我们应该改变什么？

Answer 2

当您已经发送了具有不同架构的消息时，您就有了一个主题。

这个问题可以通过多种方式解决。

### Deletes all schema versions registered under the subject "Kafka-value"
$ curl -X DELETE http://localhost:8081/subjects/Kafka-value
  [1]

### Deletes version 1 of the schema registered under subject "Kafka-value"
$ curl -X DELETE http://localhost:8081/subjects/Kafka-value/versions/1
  1

### Deletes the most recently registered schema under subject "Kafka-value"
$ curl -X DELETE http://localhost:8081/subjects/Kafka-value/versions/latest

上述 API 主要用于开发环境，在这种环境中，在最终确定架构之前通常会进行迭代。虽然不建议在生产环境中使用这些 API，但很少有场景可以在生产环境中使用这些 API，但要格外小心。

• 要注册的新架构与现有架构版本之一存在兼容性问题

• 需要为同一主题重新注册旧版本的架构

• 架构仅用于实时流系统，并且绝对不再需要旧版本

• 一个话题需要回收

还需要注意的是，在使用“删除主题”或删除唯一可用的架构版本时，任何已注册的主题兼容性设置也将被删除。

第二种方法是开始向新主题发送消息。按照这些步骤，你会没事的。

• 更新生产者以将数据发送到新主题，该主题将在架构注册表中注册更新的架构
• 验证该主题的所有消费者的延迟为零
• 更新消费者以使用来自新主题的数据
• 从 kafka 中删除旧主题

Answer 3

根据@Gary Russell 的反馈，我对我的代码进行了以下更改以使其正常工作。谢谢@Gary 和@Manjeet/@cricket_007

所以我做了以下，它工作了

注释为@KafkaListener的方法基本没有变化，但是包含它的类必须实现这些接口

MyKafkaListenerClass implements ConsumerSeekAware.ConsumerSeekCallback,ConsumerSeekAware

然后在那个类中我实现这些接口的方法......

    private final ThreadLocal<ConsumerSeekCallback> seekCallBack = new ThreadLocal<>();

@Override
public void registerSeekCallback(ConsumerSeekCallback consumerSeekCallback) {
    this.seekCallBack.set(consumerSeekCallback);
}

@Override
public void onPartitionsAssigned(Map<TopicPartition, Long> map, ConsumerSeekCallback consumerSeekCallback) {
    this.seekCallBack.get().seekToEnd(topic,0);
}

@Override
public void onIdleContainer(Map<TopicPartition, Long> map, ConsumerSeekCallback consumerSeekCallback) {

}

@Override
public void seek(String topic, int partition, long offset) {
    System.out.println("seekToEnd is hit for topic= " + topic + " and partition=" + partition+ " and offset =" + offset);
}

@Override
public void seekToBeginning(String s, int i) {

}

@Override
public void seekToEnd(String topic, int partition) {
    System.out.println("seekToEnd is hit for topic s = " + topic + " and partition i=" + partition);
}