Lost message from the Kafka Topic

Question

While trying timestamp in ProducerRecord; I found something weird. After sending few messages from the producer, I ran kafka-console-consumer.sh and verified that those messages are in the topic. I stopped the producer and waited for a minute. When I reran kafka-console-consumer.sh then it did not show the messages that I generated previously. I also added producer.flush() and producer.close() but the outcome was still the same.

Now, when I stopped using timestamp field then everything worked fine which makes me believe that there is something finicky about messages with timestamp.

I am using Kafka_2.11-2.0.0 (released on July 30, 2018)

Following is the sample code.

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.header.internal.RecordHeaders;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;
import static java.lang.Thread.sleep;
public class KafkaProducerSample{
    public static void main(String[] args){
        String kafkaHost="sample:port";
        String notificationTopic="test";

        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, this.kafkaHost);
        props.put(ProducerConfig.ACKS_CONFIG, 1);
        props.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);

        Producer<String, String> producer = new KafkaProducer(props, new StringSerialize(), new StringSerializer);

        RecordHeaders recordHeaders = new RecordHeader();
        ProducerRecord<String, String> record = new ProducerRecord(notificationTopic, null, 1574443515L, sampleKey, SampleValue);
        producer.send(record);
        sleep(1000);
    }
}

I run console consumer as following

$KAFKA_HOME/bin/kafka-console-consumer.sh --bootstrap.server KAFKA_HOST:PORT --topic test --from-beginning

#output after running producer
test


#output 5mins after shutting down producer

How did you run your kafka-console-consumer? Have you used --from-beginning flag? — Giorgos Myrianthous
try to consume from partition and specific offset bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic mytopic --offset 0 --partition 0 — Nitin

OneCricketeer OneCricketeer · Accepted Answer · 2019-11-22T22:23:26

You are asynchronously sending only one record, but not ack-ing or flushing the buffer.

You will need to send more records,

or

producer.send(record).get();

or

producer.send(record);
producer.flush();

or (preferred), do Runtime.addShutdownHook() in your main method to flush and close the producer

Lost message from the Kafka Topic

1 Answers