卡夫卡生产者跳过消息

问题描述：

我正在尝试将文件中的数据写入卡夫卡主题。我的代码如下所示：卡夫卡生产者跳过消息

Properties properties = new Properties(); 
    properties.put("bootstrap.servers", <bootstrapServers>); 
    properties.put("key.serializer", StringSerializer.class.getCanonicalName()); 
    properties.put("value.serializer", StringSerializer.class.getCanonicalName()); 
    properties.put("retries",100); 
    properties.put("linger.ms",5); 
    properties.put("acks", "all"); 

    KafkaProducer<Object, String> producer = new KafkaProducer<>(properties); 

    try (BufferedReader bf = new BufferedReader(new InputStreamReader(new FileInputStream(filePath), "UTF-8"))) { 
     String line; 
     int count = 0; 
     while ((line = bf.readLine()) != null) { 
      count++; 
      producer.send(new ProducerRecord<>(topicName, line)); 
     } 
    producer.flush(); 
     Logger.log("Done producing data messages. Total no of records produced:" + count); 
    } catch (InterruptedException | ExecutionException | IOException e) { 
     Throwables.propagate(e); 
    } finally { 
     producer.close(); 
    }

数据大小高于100万条记录。

当我检查上使用下面的命令代理偏移数据的，只有消息的一半（约5,00,000）被写入在话题：

./kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list <broker_list> --time -1 --topic <topic_name>

上述命令的输出：

topic_name:1:292954 
topic_name:0:296787

我应该怎么做才能确保所有内容都写在主题上。

你能显示GetOffsetShell命令的实际输出吗？ – C4stor

在问题中添加了输出。 –

应用程序日志文件中的计数值是多少？它显示1米吗？ – notionquest

答

发送消息是异步的。在处理所有消息之前，您可能正在检查偏移量。

日志保留时间为24小时。而且我正在检查消息后产生消息，这几乎需要4-5分钟。 –

你完全改变了答案。 –

是的，我的想法很糟糕，意识到保留不会成为问题，我在更改之前忘记刷新 –

卡夫卡生产者跳过消息

相关推荐