@KafkaListener Каждый раз читать с начала

use a compacted topic with 1 partition to hold a list of configurations. This then needs to be called by a rest end point and it should display a full unique list of the configurations

То, как вы должны реализовать это, - использовать потоки Kafka и KTable и настроить интерактивные запросы за вашим уровнем REST. Не стандартный потребитель, которому нужно перематывать себя, чтобы получить наиболее актуальное состояние системы.

Пример этого уже существует в платформе Kafka Connect, где у него есть тема конфигурации, и вы можете получить доступ только к самому последнему значению GET /connectors/name/config, и только если вы перезапустите его или масштабируете до большего количества экземпляров, он снова потребляет все сообщения . Реестр схем также является примером этого и хранит внутреннюю хэш-карту всех схем в теме _schemas и имеет REST API для чтения, вставки, удаления.

По сути, когда вы получаете новую конфигурацию для данного ключа, вы можете либо «заменить» старое значение для данного ключа совершенно новым, либо каким-то образом «объединить» старое значение с новыми данными.

Я думаю, вам следует попробовать написать ConsumerSeekAware Listener и искать смещение 0 каждый раз, когда вы читаете сообщение. Звучит как безумный обходной путь, но он может помочь. Надеюсь, что это поможет вам :-)

class Listener implements ConsumerSeekAware {

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

   ----Override all methods that are needed----

@KafkaListener(...)
    public void listen(@Payload String message) {

            this.seekCallBack.get().seek(topic, partition, 0);
        }
    }
}

@ Nimo1981 Итак, это реализация на простой Java. Я не уверен, что это соответствует вашим потребностям. Итак, в основном я фиксирую смещение 0 (то есть, даже если я читаю из темы Kafka, я возвращаюсь к смещению, которое находится в начале.) Я не уверен, рассматривали ли вы эту реализацию, но, пожалуйста, дайте мне знать, если это это то, что вы ищете

Оставьте CommitCountObj. Это вам не нужно. Таким образом, по умолчанию offsetMap будет иметь следующую запись смещения, подобную этой,

offsetMap.put (новый TopicPartition (record.topic (), record.partition ()), новый OffsetAndMetadata (record.offset () + 1, «какое-то сообщение об успешной фиксации»));

но для вашего варианта использования я как бы модифицировал, он хорошо работает, когда потребитель не перезапускается

offsetMap.put (новый TopicPartition (record.topic (), record.partition ()), новый OffsetAndMetadata (0, «фиксация не выполнена»));

public class KafkaConsumerClass {

    private static final org.apache.log4j.Logger log = org.apache.log4j.Logger.getLogger(KafkaConsumerClass.class);
    private CommitCountClass commitCountobj = new CommitCountClass();

    public Consumer<String, List<FeedBackConsumerClass>> createConsumer() {
        Map<String, Object> consumerProps = new HashMap<String, Object>();
        consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:7070,localhost:7072");
        consumerProps.put(ConsumerConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG, 50000);
        consumerProps.put(ConsumerConfig.CLIENT_ID_CONFIG, "first_group-client1");
        // consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
        consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "first_group");
        // consumerProps.put(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, KafkaConsumerInterceptor.class);
        consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        consumerProps.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 15000);
        consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        consumerProps.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1500);
        consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        return new KafkaConsumer<String, List<FeedBackConsumerClass>>(consumerProps);
    }

    public void consumeRecord() {
        log.info("Coming inside consumer consumer");
        ArrayList<String> topicList = new ArrayList<String>();
        topicList.add("topic1");
        commitCountobj.setCount(0);
        Consumer<String, List<FeedBackConsumerClass>> kafkaConsumer = createConsumer();
        kafkaConsumer.subscribe(topicList);
        log.info("after subscribing");

        Map<TopicPartition, OffsetAndMetadata> offsetMap = new HashMap<>();

        while (true) {

            ConsumerRecords<String, List<FeedBackConsumerClass>> recordList = kafkaConsumer.poll(Long.MAX_VALUE);
            // kafkaConsumer.seekToBeginning(kafkaConsumer.assignment());

            log.info("Inside while loop:" + recordList);
            if (!recordList.isEmpty()) {
                recordList.forEach(record -> {
                    int i = 0;
                    System.out.println(record.toString());
                    // we can make the call to the API here
                    // call the db here or any API and process the record
                    // then call the code to commit
                    // since the commit is switched off, it becomes a developers responsibility to do the auto commit
                    offsetMap.put(new TopicPartition(record.topic(), record.partition()),
                            new OffsetAndMetadata(0, "no metadata/offset commited"));
                    // here we are incrementing the offsetMap so that we are making sure we are storing the
                    // next set of offsets in the map
                    if (commitCountobj.getCount() % 1000 == 0) {
                        kafkaConsumer.commitAsync(offsetMap, new OffsetCommitCallback() {

                            @Override
                            public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets,
                                    Exception exception) {
                                // TODO Auto-generated method stub
                                if (exception != null) {
                                    // retry it now with a sync
                                    // possibility of error occuring here as well
                                    // so capture the exception and exit the consumer gracefully
                                    kafkaConsumer.commitSync();
                                    log.error(exception.getMessage());
                                }
                            }
                        });
                    }
                    commitCountobj.setCount(i++);
                });
            }

        }
    }

}

Вот как я это буду реализовывать. Вам необходимо реализовать интерфейс ConsumerSeekAware и выполнить некоторые реализации метода onPartitionsAssigned. Вы также можете сделать seekToBegining по запросу, если отправляете переменную среды при перезапуске приложения. Но я не реализовал это!

@Service
@EnableKafka
public class Service implements ConsumerSeekAware {



    @KafkaListener(topics = "${topicName}", groupId = "${groupId}")
    public void listen(@Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
                       @Header(KafkaHeaders.RECEIVED_TIMESTAMP) long ts,
                       @Payload List<String> messageBatch
    ) {
            //do a bunch of stuff
    }



    @Override
    public void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
        String topic= Optional.ofNullable(System.getProperty(TOPIC_NAME)).orElseThrow(()->new RuntimeException("topicName needs to be set"));
        assignments.keySet().stream().filter(partition->topic.equals(partition.topic()))
                .forEach(partition -> callback.seekToBeginning(topic, partition.partition()));
    }

    @Override
    public void onIdleContainer(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {}

    @Override
    public void registerSeekCallback(ConsumerSeekCallback callback) {}
}

@KafkaListener(topicPartitions 
          = @TopicPartition(topic = "test", partitionOffsets = {
          @PartitionOffset(partition = "0", initialOffset = "0")}),groupId = "foo",
        containerFactory = "kafkaListenerContainerFactory")
public void listenAllMsg(@Payload String message,@Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition) {
    System.out.println(" all msg Received Messasge in group 'foo': " + message+"RECEIVED_PARTITION_ID - "+partition);

}

в кафке 2.3.1