对不起，我的错。在轮询时，我没有为“无现有偏移量”设置阻止符。所以下面的代码在我设置了重试次数的上限之后运行得很好，可以随意发表评论了，但是我想知道使用int来停止这样的重试是否是最好的做法？谢谢

props.put("enable.auto.commit", "false");
        props.put("max.poll.records", "1");
        props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "appname-message-patching");

        List<Pair<Integer,Integer>> partition_offset= Arrays.asList(Pair.of(0,1),Pair.of(0,21),
                Pair.of(1,31) );

        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);

        partition_offset.forEach(po-> {
            TopicPartition tp=new TopicPartition(topicName, po.getKey());
            consumer.assign(Arrays.asList(tp));
            consumer.seek(tp,po.getValue());
            int counter=1;//only get 1 message , double guards for max.poll.records ?
            int retry=5;//retry 5 times then give up if cannot receive anything
            while (counter>0 ) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));

                if(retry>0 || !records.isEmpty()) {
                    for (ConsumerRecord<String, String> record : records) {
                        counter--;
                        System.out.printf("==================thread name = %s , partation = %d , offset = %d , key = %s , value = %s\n", Thread.currentThread().getName(), record.partition(), record.offset(), record.key(), record.value());
                    }
                    retry--;
                }
                if (retry==0||counter==0){
                    counter=0;break;
                }
            }
        });
        consumer.close();