Handled isEmpty operation on dataframe, logging enhancements and fixes to checkpoint handling

Author: SravanThotakura05

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

@@ -201,20 +201,23 @@ public Offset initialOffset() {
         if(existingCheckpoints != null && !existingCheckpoints.isEmpty()) {
             currentCheckpoint = existingCheckpoints;
             existingCheckpoints.forEach(checkpoint -> lastKnownMessageIds = String.join(",", lastKnownMessageIds, checkpoint.getMessageIDs()));
-
+            log.info("SolaceSparkConnector - Checkpoint available from LVQ {}", new Gson().toJson(existingCheckpoints));
             return new SolaceSourceOffset(lastKnownOffsetId, existingCheckpoints);
         }
-
+        log.info("SolaceSparkConnector - Initial Offset from LVQ is not available, the micro integration will use the available offset in checkpoint else a new checkpoint state will be created");
         return new SolaceSourceOffset(lastKnownOffsetId, new CopyOnWriteArrayList<>());
     }
 
     @Override
     public Offset deserializeOffset(String json) {
         SolaceSourceOffset solaceSourceOffset = getDeserializedOffset(json);
-        if(solaceSourceOffset != null) {
-            lastKnownOffsetId = solaceSourceOffset.getOffset();
-            solaceSourceOffset.getCheckpoints().forEach(checkpoint -> lastKnownMessageIds = String.join(",", lastKnownMessageIds, checkpoint.getMessageIDs()));
+        if(solaceSourceOffset.getCheckpoints() != null && solaceSourceOffset.getCheckpoints().isEmpty()) {
+            log.info("SolaceSparkConnector - No offset is available in spark checkpoint location. New checkpoint state will be created");
+        } else {
+            log.info("SolaceSparkConnector - Deserialized offset {}", new Gson().toJson(solaceSourceOffset));
         }
+        lastKnownOffsetId = solaceSourceOffset.getOffset();
+        solaceSourceOffset.getCheckpoints().forEach(checkpoint -> lastKnownMessageIds = String.join(",", lastKnownMessageIds, checkpoint.getMessageIDs()));
 
         return solaceSourceOffset;
     }
@@ -229,6 +232,8 @@ private SolaceSourceOffset getDeserializedOffset(String json) {
                 } else {
                     return migrate(solaceSourceOffset.getOffset(), "");
                 }
+            } else {
+                return solaceSourceOffset;
             }
         } catch (Exception e) {
             log.warn("SolaceSparkConnector - Exception when deserializing offset. May be due incompatible formats. Connector will try to migrate to latest offset format.");
@@ -244,8 +249,6 @@ private SolaceSourceOffset getDeserializedOffset(String json) {
                 throw new RuntimeException("SolaceSparkConnector - Exception when migrating offset to latest format.", e);
             }
         }
-
-        return null;
     }
 
     private SolaceSourceOffset migrate(int offset, String messageIds) {

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

@@ -11,7 +11,7 @@
 import java.util.concurrent.CopyOnWriteArrayList;
 
 public final class SolaceMessageTracker implements Serializable {
-    private static String lastBatchId = "";
+    private static ConcurrentHashMap<String, String> lastBatchId = new ConcurrentHashMap<>();
     private static final Logger logger = LogManager.getLogger(SolaceMessageTracker.class);
     private static ConcurrentHashMap<String, CopyOnWriteArrayList<SolaceMessage>> messages = new ConcurrentHashMap<>();
     private static ConcurrentHashMap<String, String> lastProcessedMessageId = new ConcurrentHashMap<>();
@@ -77,11 +77,11 @@ public static void resetId(String uniqueId) {
         logger.info("SolaceSparkConnector - Cleared all messages from Offset Manager for {}", uniqueId);
     }
 
-    public static String getLastBatchId() {
-        return lastBatchId;
+    public static String getLastBatchId(String uniqueId) {
+        return lastBatchId.get(uniqueId);
     }
 
-    public static void setLastBatchId(String lastBatchId) {
-        SolaceMessageTracker.lastBatchId = lastBatchId;
+    public static void setLastBatchId(String uniqueId, String batchId) {
+        lastBatchId.put(uniqueId, batchId);
     }
 }

src/main/java/com/solacecoe/connectors/spark/streaming/partitions/SolaceInputPartitionReader.java

@@ -72,7 +72,7 @@ public SolaceInputPartitionReader(SolaceInputPartition inputPartition, boolean i
          * In case when multiple operations are performed on dataframe, input partition will be called as part of Spark scan.
          * We need to acknowledge messages only if new batch is started. In case of same batch we will return the same messages.
          */
-        if (!currentBatchId.equals(SolaceMessageTracker.getLastBatchId())) {
+        if (!currentBatchId.equals(SolaceMessageTracker.getLastBatchId(this.uniqueId))) {
             /* Currently solace can ack messages on consumer flow. So ack previous messages before starting to process new ones.
              * If Spark starts new input partition it indicates previous batch of data is successful. So we can acknowledge messages here.
              * Solace connection is always active and acknowledgements should be successful. It might throw exception if connection is lost
@@ -90,7 +90,7 @@ public SolaceInputPartitionReader(SolaceInputPartition inputPartition, boolean i
             }
         }
 
-        SolaceMessageTracker.setLastBatchId(currentBatchId);
+        SolaceMessageTracker.setLastBatchId(this.uniqueId, currentBatchId);
 
         this.includeHeaders = includeHeaders;
         this.properties = properties;
@@ -239,7 +239,7 @@ private SolaceMessage getNextMessage() {
             while (shouldProcessMoreMessages(batchSize, messages)) {
                 try {
                     if (iterator.hasNext()) {
-                        shouldTrackMessage = false;
+//                        shouldTrackMessage = false;
                         solaceMessage = iterator.next();
                         if (solaceMessage == null) {
                             return null;

src/main/java/com/solacecoe/connectors/spark/streaming/solace/EventListener.java

@@ -48,10 +48,7 @@ public void onReceive(BytesXMLMessage msg) {
             if(!this.checkpoints.isEmpty()) {
                 List<String> lastKnownMessageIDs = new ArrayList<>();
                 String messageID = SolaceUtils.getMessageID(msg, this.offsetIndicator);
-                boolean hasPartitionKey = false;
-                if(msg.getProperties() != null && msg.getProperties().containsKey(XMLMessage.MessageUserPropertyConstants.QUEUE_PARTITION_KEY)) {
-                    hasPartitionKey = true;
-                }
+                boolean hasPartitionKey = msg.getProperties() != null && msg.getProperties().containsKey(XMLMessage.MessageUserPropertyConstants.QUEUE_PARTITION_KEY);
                 if(!hasPartitionKey) {
                     for(SolaceSparkPartitionCheckpoint checkpoint : this.checkpoints) {
                         lastKnownMessageIDs.addAll(Arrays.stream(checkpoint.getMessageIDs().split(",")).collect(Collectors.toList()));
@@ -87,7 +84,7 @@ public void onReceive(BytesXMLMessage msg) {
                             this.messages.add(new SolaceMessage(msg));
                         }
                     } else {
-                        log.warn("SolaceSparkConnector - Incoming message has no partition key but messages in checkpoint has partition id's. Message will be reprocessed to ensure reliability—any duplicates must be handled by the downstream system.");
+                        log.warn("SolaceSparkConnector - Incoming message partition key is either null or empty. Message will be reprocessed to ensure reliability—any duplicates must be handled by the downstream system.");
                         this.messages.add(new SolaceMessage(msg));
                     }
                 }
@@ -112,7 +109,7 @@ private void compareMessageIds(List<String> lastKnownMessageIDs, String messageI
             } else {
                 this.messages.add(new SolaceMessage(msg));
                 if (lastKnownMessageIDs.size() > 1) {
-                    log.warn("SolaceSparkConnector - Message ID {} not acknowledged as it may be a previously failed message received before newer checkpointed ones(parallel consumers). The connector will not check against checkpointed message IDs. It will be reprocessed to ensure reliability—any duplicates must be handled by the downstream system.", messageID);
+                    log.warn("SolaceSparkConnector - Message ID {} was not acknowledged because it may be an older, previously failed message received ahead of more recent checkpointed messages (due to parallel consumers). The connector does not validate message IDs against checkpoints, so this message will be reprocessed to ensure reliability. Any resulting duplicates must be handled by the downstream system.”", messageID);
                     break;
                 }
             }

src/main/java/com/solacecoe/connectors/spark/streaming/solace/SolaceBroker.java

@@ -210,18 +210,35 @@ public void createLVQIfNotExist() {
 
         EndpointProperties endpoint_props = new EndpointProperties();
         endpoint_props.setAccessType(EndpointProperties.ACCESSTYPE_EXCLUSIVE);
-        endpoint_props.setPermission(EndpointProperties.PERMISSION_CONSUME);
         endpoint_props.setQuota(0);
         try {
             this.session.provision(lvq, endpoint_props, JCSMPSession.FLAG_IGNORE_ALREADY_EXISTS | JCSMPSession.WAIT_FOR_CONFIRM);
         } catch (JCSMPException e) {
             log.error("SolaceSparkConnector - Exception creating LVQ. Shutting down consumer ", e);
             close();
+            this.isException = true;
+            this.exception = e;
+            throw new RuntimeException(e);
         }
         try {
             this.session.addSubscription(lvq, JCSMPFactory.onlyInstance().createTopic(this.lvqTopic), JCSMPSession.WAIT_FOR_CONFIRM);
         } catch (JCSMPException e) {
-            log.warn("SolaceSparkConnector - Subscription already exists on LVQ. Ignoring error");
+            if(e instanceof JCSMPErrorResponseException) {
+                JCSMPErrorResponseException jce = (JCSMPErrorResponseException) e;
+                if(jce.getResponsePhrase().contains("Subscription Already Exists")) {
+                    log.warn("SolaceSparkConnector - Subscription Already Exists on LVQ {}", this.lvqName);
+                } else {
+                    close();
+                    this.isException = true;
+                    this.exception = e;
+                    throw new RuntimeException(e);
+                }
+            } else {
+                close();
+                this.isException = true;
+                this.exception = e;
+                throw new RuntimeException(e);
+            }
         }
     }
 
@@ -233,10 +250,10 @@ public CopyOnWriteArrayList<SolaceSparkPartitionCheckpoint> browseLVQ() {
         try {
             Browser myBrowser = session.createBrowser(br_prop);
             BytesXMLMessage rx_msg = null;
-            log.info("SolaceSparkConnector - Browsing message from LVQ {}", this.lvqName);
+            log.info("SolaceSparkConnector - Browsing checkpoint from LVQ {}", this.lvqName);
             rx_msg = myBrowser.getNext();
             if (rx_msg != null) {
-                log.info("SolaceSparkConnector - Browsed message from LVQ {}", this.lvqName);
+                log.info("SolaceSparkConnector - Browsed checkpoint from LVQ {}", this.lvqName);
                 byte[] msgData = new byte[0];
                 if (rx_msg.getContentLength() != 0) {
                     msgData = rx_msg.getBytes();
@@ -247,7 +264,7 @@ public CopyOnWriteArrayList<SolaceSparkPartitionCheckpoint> browseLVQ() {
                 lastKnownCheckpoint = new Gson().fromJson(new String(msgData, StandardCharsets.UTF_8), new TypeToken<CopyOnWriteArrayList<SolaceSparkPartitionCheckpoint>>(){}.getType());
                 log.info("SolaceSparkConnector - Checkpoint from LVQ {}", new String(msgData, StandardCharsets.UTF_8));
             } else {
-                log.info("SolaceSparkConnector - No message available from LVQ {}", this.lvqName);
+                log.info("SolaceSparkConnector - No checkpoint available in LVQ {}", this.lvqName);
             }
             myBrowser.close();
             return lastKnownCheckpoint;