title: Collecting messages which are not successfully written on Kafka
body: <p>I am reading a file and dumping each record on Kafka. Here is my producer code:</p>
  
  <pre><code>public void produce(String topicName, String filePath, String bootstrapServers, String encoding) {
       try (BufferedReader bf = getBufferedReader(filePath, encoding);
                   KafkaProducer&lt;Object, String&gt; producer = initKafkaProducer(bootstrapServers)) {
                  String line;
                  long count = 0;
                  while ((line = bf.readLine()) != null) {
                      count++;
                      producer.send(new ProducerRecord&lt;&gt;(topicName, line), (metadata, e) -&gt; {
                          if(e != null){
                              e.printStackTrace();
                              //write record to some file.
                          }
                      });
                  }
                  producer.flush();
                  CustomLogger.log("Done producing data messages. Total no of records produced:" + count);
              } catch (IOException e) {
                  Throwables.propagate(e);
              }
  }
   private static KafkaProducer&lt;Object, String&gt; initKafkaProducer(String bootstrapServer) {
          Properties properties = new Properties();
          properties.put("bootstrap.servers", bootstrapServer);
          properties.put("key.serializer", StringSerializer.class.getCanonicalName());
          properties.put("value.serializer", StringSerializer.class.getCanonicalName());
          properties.put("acks", "-1");
          properties.put("retries", 4);
          return new KafkaProducer&lt;&gt;(properties);
      }
  
  private BufferedReader getBufferedReader(String filePath, String encoding) throws UnsupportedEncodingException, FileNotFoundException {
      return new BufferedReader(new InputStreamReader(new FileInputStream(filePath), Optional.ofNullable(encoding).orElse("UTF-8")));
  }
  </code></pre>
  
  <p>As per our basic tests there are chances that producing message can fail due to TimeoutException. However as per the <a href="https://kafka.apache.org/090/javadoc/org/apache/kafka/clients/producer/Callback.html" rel="nofollow noreferrer">official documentation of Callback</a> TimeoutException is a retriable exception. Means in next retries this message might get produced. Hence I can not consider that record sending is failed if I find TimeoutException in callback. Is there any realiable way using which I can surely say that record sending is failed and log that in separate file?</p>