Check for exception before starting input partition

Author: SravanThotakura05

src/main/java/com/solacecoe/connectors/spark/streaming/SolaceMicroBatch.java

@@ -106,6 +106,7 @@ public Offset latestOffset() {
 
     @Override
     public InputPartition[] planInputPartitions(Offset start, Offset end) {
+        checkException();
         if(partitions == 0) {
             partitions = getTotalExecutors();
         }
@@ -275,6 +276,7 @@ public void commit(Offset end) {
         if(!lastKnownOffset.isEmpty()) {
             log.trace("SolaceSparkConnector - Final checkpoint publishing to LVQ {}", new Gson().toJson(this.lastKnownOffset));
             this.solaceBroker.publishMessage(properties.getOrDefault(SolaceSparkStreamingProperties.SOLACE_SPARK_CONNECTOR_LVQ_TOPIC, SolaceSparkStreamingProperties.SOLACE_SPARK_CONNECTOR_LVQ_DEFAULT_TOPIC), new Gson().toJson(this.lastKnownOffset));
+            checkException();
             lastKnownOffset.clear();
         }
     }
@@ -289,6 +291,12 @@ private CopyOnWriteArrayList<SolaceSparkPartitionCheckpoint> getCheckpoint() {
         return this.solaceBroker.getOffsetFromLvq();
     }
 
+    private void checkException() {
+        if(this.solaceBroker.isException()) {
+            throw new RuntimeException(this.solaceBroker.getException());
+        }
+    }
+
     private String convertCheckpointURIToStringPath(String checkpointLocation) {
         if (checkpointLocation.startsWith("dbfs:/")) {
             // Strip "dbfs:/" and prepend "/dbfs/"

src/main/java/com/solacecoe/connectors/spark/streaming/offset/SolaceMessageTracker.java

@@ -40,8 +40,8 @@ public static void addMessage(String uniqueId, SolaceMessage message) {
 
     public static void ackMessages(String uniqueId) {
         if(messages.containsKey(uniqueId)) {
-            logger.trace("SolaceSparkConnector - Acknowledging {} messages ", messages.get(uniqueId).size());
             messages.get(uniqueId).forEach(message -> message.bytesXMLMessage.ackMessage());
+            logger.trace("SolaceSparkConnector - Acknowledged {} messages ", messages.get(uniqueId).size());
             messages.remove(uniqueId);
         }
     }