[GOBBLIN-2044] Catch and log exceptions in HighLevelConsumer queue co…

…nsumption (#3923)

Co-authored-by: Urmi Mustafi <umustafi@linkedin.com>

gobblin-runtime/src/main/java/org/apache/gobblin/runtime/kafka/HighLevelConsumer.java

@@ -335,9 +335,10 @@ public QueueProcessor(BlockingQueue queue) {
     @Override
     public void run() {
       log.info("Starting queue processing.. " + Thread.currentThread().getName());
+      KafkaConsumerRecord record = null;
       try {
         while (true) {
-          KafkaConsumerRecord record = queue.take();
+          record = queue.take();
           messagesRead.inc();
           HighLevelConsumer.this.processMessage((DecodeableKafkaRecord)record);
           recordsProcessed.incrementAndGet();
@@ -349,9 +350,11 @@ public void run() {
           }
         }
       } catch (InterruptedException e) {
-        log.warn("Encountered exception while processing queue ", e);
+        log.warn("Thread interrupted while processing queue ", e);
         // TODO: evaluate whether we should interrupt the thread or continue processing
         Thread.currentThread().interrupt();
+      } catch (Exception e) {
+        log.error("Encountered exception while processing record so stopping queue processing. Record: {} Exception: {}", record, e);
       }
     }
   }