Kafka主题中的丢失消息

问题描述 投票:1回答:1

[尝试使用ProducerRecord中的时间戳记;我发现了一些奇怪的东西。在从生产者发送少量消息之后,我运行了kafka-console-consumer.sh并验证了这些消息是否在主题中。我停下了制片人,等了一分钟。当我重新运行kafka-console-consumer.sh时,它没有显示我先前生成的消息。我还添加了producer.flush()和producer.close(),但结果仍然相同。

现在,当我停止使用时间戳字段时,一切工作正常,这使我相信带有时间戳的消息有些挑剔。

我正在使用Kafka_2.11-2.0.0(于2018年7月30日发布)

下面是示例代码。

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);
    }
}

我如下运行控制台使用者

$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
java apache-kafka kafka-producer-api
1个回答
2
投票

您仅异步发送一条记录,但没有确认或刷新缓冲区。

您将需要发送更多记录,

producer.send(record).get();

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

或(首选),在您的主要方法中执行Runtime.addShutdownHook()刷新并关闭生产者

© www.soinside.com 2019 - 2024. All rights reserved.