Page History

{

// …

  try (KafkaConsumer<String, SimpleValue> consumer = new KafkaConsumer<>(settings())) {

     // Subscribe to our topic
     LOGGER.info("Subscribing to topic " + KAFKA_TOPIC);
     consumer.subscribe(List.of(KAFKA_TOPIC));
     LOGGER.info("Subscribed !");

     try (KafkaProducer<byte[], byte[]> dlqProducer = new KafkaProducer<>(producerSettings())) {
         //noinspection InfiniteLoopStatement
         while (true) {
             try {
                 final var records = consumer.poll(POLL_TIMEOUT);
                 LOGGER.info("poll() returned {} records", records.count());
                 for (var record : records) {
                     LOGGER.info("Fetch record key={} value={}", record.key(), record.value());
                     // Any processing
                     // ...
                 }

             } catch (RecordDeserializationException re) {
                 long offset = re.offset();
                 Throwable t = re.getCause();
                 LOGGER.error("Failed to consume at partition={} offset={}", re.topicPartition().partition(), offset, t);
                 sendDlqRecord(dlqProducer, re());
                 LOGGER.info("Skipping offset={}", offset);
                 consumer.seek(re.topicPartition(), offset + 1);
             } catch (Exception e) {
                 LOGGER.error("Failed to consume", e);
             }
         }
     }

  } finally {
     LOGGER.info("Closing consumer");
  }
}


void sendDlqRecord(KafkaProducer<byte[], byte[]> dlqProducer, RecordDeserializationException re) {
   var dlqRecord = new ProducerRecord<>(DLQ_TOPIC, re.key(), re.value());

   try {
       dlqProducer.send(dlqRecord).get();
       LOGGER.info("Record sent to DLQ");
   } catch (Exception e) {
       LOGGER.error("Failed to send corrupted record to DLQ", e);
   }
}