Apache Kafka的Spring：如何搜索分区的ToEnd？

Question

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!

我已经更改了包含@KafKalistener方法的@Service类，以实现文档中提到的接口

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

它有@KafKalistener注释的方法，我曾尝试在该方法中搜索分区的ToEnd

@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);
}

    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!

我正在使用这里的Spring Kafka模板实现示例的片段，用于搜索偏移量、确认

正如这里提到的，是什么决定了Kafka消费者偏移量？，我不能使用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的参数。

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