Sou novo na Kafka e estou trabalhando em um protótipo para conectar um serviço de streaming proprietário ao Kafka.

Estou procurando obter a chave da última mensagem em um tópico, pois nosso consumidor interno de fluxo precisa fazer logon com o ID da última mensagem que recebeu ao se conectar.

É possível, usando o KafkaProducer ou um KafkaConsumer para fazer isso?

Tentei fazer o seguinte usando um consumidor, mas ao executar o consumidor do console, vejo as mensagens repetidas.

    // Poll so we know we're connected
    consumer.poll(100);
    // Get the assigned partitions
    Set<TopicPartition> assignedPartitions = consumer.assignment();
    // Seek to the end of those partitions
    consumer.seekToEnd(assignedPartitions);

    for(TopicPartition partition : assignedPartitions) {
        final long offset = consumer.committed(partition).offset();
        // Seek to the previous message
        consumer.seek(partition,offset - 1);
    }

    // Now get the last message
    ConsumerRecords<String, String> records = consumer.poll(100);
    for (ConsumerRecord<String, String> record : records) {
        lastKey = record.key();
    }
    consumer.close();

Esse comportamento é esperado ou estou no caminho errado?