kafka异常处理

2c
1 parent 28f7c958
Showing 1 changed file with 9 additions and 0 deletions
trash-workFlow/src/main/java/com/trash/casefile/kafka/Consumer.java
@@ -42,6 +42,15 @@ public class Consumer {
     @KafkaListener(topics = "record_process_alarm")
     public void consume(ConsumerRecord<?, ?> record, Acknowledgment ack) throws InterruptedException, IOException {
+        if(record==null){
+            throw new RuntimeException("kafka record对象为空");
+        }
+        if(record.value()==null){
+            throw new RuntimeException("kafka消费数据为空");
+        }
+        if(ack==null){
+            throw new RuntimeException("kafka ack对象为空");
+        }
         log.info("kafka消费数据成功,offset:"+ record.offset() +",data:" + record.value());
         String id = insertKafkaCompensation(record.value().toString());
         autoViolationWarning(record.value().toString(),id);