我用的是Spring Boot 2.0.2.RELEASE 使用spring for ApacheKafka（有效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 建议实现ConsumerSekAware及其子接口ConsumerSekAware.ConsumerSekCallback
我已经更改了包含@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!

我在这里使用SpringKafka模板实现示例中的片段来查找偏移量、确认
正如这里提到的，什么决定了kafka consumer offset？我不能使用auto.offset.reset属性从主题末尾开始消费（除非我使用不同的consumergroupid-在我的情况下这是不可能的）。我想知道我是否可以利用现有的消费者群体来解决这个问题。