Почему мой потребитель Java не прочитает данные, которые я создал?

Question

Я пытаюсь прочитать данные от простого созданного мной производителя.По какой-то причине, когда я запускаю потребителя, он не видит и не производит никаких данных, которые я произвел.Может ли кто-нибудь дать мне какие-либо рекомендации о том, что делать дальше?

Я включил код моего производителя и потребителя ниже:

Производитель:

public class AvroProducer {

public static void main(String[] args) {

    String bootstrapServers = "localhost:9092";
    String topic = "trackingReportsReceived";

    //create Producer properties
    Properties properties = new Properties();
    properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class.getName());
    properties.setProperty("schema.registry.url", "http://localhost:8081");

    //create the producer
    KafkaProducer<String, trackingReport> producer = new KafkaProducer<>(properties);

    //creating my own event
    trackingReport event = trackingReport.newBuilder()
            .setRemoteEventUID(2)
            .setReceivedPacketUID(2)
            .setRemoteUnitAID(2)
            .setEventTime(2)
            .setEventLocationStampUID(3)
            .setEventLatitude(2)
            .setEventLongitude(2)
            .setEventOdometer(3)
            .setEventSpeed(3)
            .setEventCourse(3)
            .build();


    //create a producer record
    ProducerRecord<String, trackingReport> eventRecord = new ProducerRecord<>(topic, event);

    //send data - asynchronous
    producer.send(eventRecord, new Callback() {
        @Override
        public void onCompletion(RecordMetadata recordMetadata, Exception e) {
            if (e == null) {
                    System.out.println("Success!");
                    System.out.println(recordMetadata.toString());
                } else {
                    e.printStackTrace();
                }
            }
        });


    //flush data
    producer.flush();
    //flush and close producer
    producer.close();

Потребитель:

public class AvroConsumer {

public static void main(String[] args) {

    final Logger logger = LoggerFactory.getLogger(AvroConsumer.class);

    String bootstrapServers = "localhost:9092";

    //create Consumer properties
    Properties properties = new Properties();
    properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    properties.put(ConsumerConfig.GROUP_ID_CONFIG, "consumer");
    properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
    properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

//        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
    properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class.getName());
    properties.put("schema.registry.url", "http://localhost:8081");
    properties.put("specific.avro.reader", "true");

    //create the consumer
    KafkaConsumer<String, trackingReport> consumer = new KafkaConsumer<>(properties);
    String topic = "trackingReportsReceived";

    consumer.subscribe(Collections.singletonList(topic));

    System.out.println("Waiting for data...");

//        try {

        while (true) {
            ConsumerRecords<String, trackingReport> records = consumer.poll(100);
            for (ConsumerRecord<String, trackingReport> record : records) {
                trackingReport trackingrep = record.value();
                System.out.println(trackingrep);
            }
            consumer.commitSync();
        }

//        } catch (Exception e) {
//            logger.error("Exception occured while consuming messages...", e);
//        } finally {
//            consumer.close();
//        }


}
}

NB. Производитель работает, а потребитель - нет.

Answer 1

С помощью сценария консольного потребителя, вы использовали тот же идентификатор группы, что и в вашем Java-потребителе?

Если вы сделали это для проверки, попробуйте использовать новую группу потребителей в вашем коде.

Если это работает, то это будет означать, что потребитель консоли прочитал все данные в теме, поэтому потребитель с этим идентификатором группы фиксирует последние текущие смещения, и когда вы запускаете Java-потребителя с тем же идентификатором группы, он пытался прочитать вэто смещение, которое является последним .. Так что никаких сообщений для чтения.

Чтобы проверить, что вы также можете сначала запустить потребитель Java, а после этого запустить производителя, если вы видите сообщения, то это будет то, что консольи Java-потребитель имел одинаковый идентификатор группы.

Answer 2

Если вы запустите потребителя после того, как он произвел записи, он не получит их.Так как, если для группы ранее не было смещений, потребитель начинает с конечных тем.

Может быть в вашем kafka-console-consumer.sh, у вас есть флаг --from-begin, который заставляетпотребитель должен начать читать с начала темы.

Вы можете явно использовать seekToBeginning (), когда ваш потребитель начинает перемещать свою позицию в начало темы.