Enhancements to LVQ Creation, handling operations on multiple data frame and improvements to ack handling, logging & documentation

Author: SravanThotakura05

src/docs/asciidoc/User-Guide.adoc

@@ -35,7 +35,7 @@ This guide assumes you are familiar with Spark set up and Spark Structured Strea
 
 === Supported Platforms
 
-The connector is built on the Spark Structured Streaming API and has been tested on Azure Databricks(15.4 LTS (includes Apache Spark 3.5.0, Scala 2.12) with photon acceleration disabled). Since the Databricks runtime is consistent across all supported cloud platforms(AWS & Google Cloud), it is expected to behave similarly in other Databricks environments. Additionally, the connector has been validated on vanilla Apache Spark, ensuring compatibility with any platform that supports standard Spark deployments.
+The connector is built on the Spark Structured Streaming API and has been tested on Azure Databricks(15.4 LTS (includes Apache Spark 3.5.0, Scala 2.12) with photon acceleration disabled and 16.4 LTS(includes Apache Spark 3.5.2, Scala 2.12)). Since the Databricks runtime is consistent across all supported cloud platforms(AWS & Google Cloud), it is expected to behave similarly in other Databricks environments. Additionally, the connector has been validated on vanilla Apache Spark, ensuring compatibility with any platform that supports standard Spark deployments.
 
 === Quick Start common steps
 
@@ -67,12 +67,27 @@ Solace Spark connector relies on Spark Checkpointing mechanism to resume from la
 
 === Checkpoint Handling
 
-Starting from version 3.1.0 connector, solace connection is now executed on worker node instead of driver node. This give us the ability to utilize cluster resource efficiently and also improves processing performance. The connector uses Solace LVQ to store checkpoint along with Spark Checkpoint.
+Starting from version 3.1.0 connector, solace connection is now executed on worker node instead of driver node. This gives us the ability to utilize cluster resource efficiently and also improves processing performance. The connector uses Solace LVQ to store checkpoint along with Spark Checkpoint.
 
 NOTE: In case of recovery, connector uses offset state from LVQ to identify last successfully processed messages. Hence, it is recommended not to delete or modify offset state in LVQ.
 
 In some cases, there might be checkpoint failures as spark may fail to write to checkpoint during instance crash or unavailability or other reasons. Though the connector will handle duplicates in most cases, we recommend to keep your downstream systems idempotent.
 
+==== Pre-requisites for LVQ creation
+The following pre-requisites are applicable for LVQ that are provisioned by Solace Administrator.
+1. The Queue should be of type Exclusive
+2. Spool Quota should be set to 0
+3. Owner of the Queue should be the client username used by the micro integration
+4. Non-Owner access should be set to No Access to prevent unauthorized access
+5. Add a topic subscription
+6. Ensure the ACL applied to the user has publish access to the topic subscribed by LVQ
+
+The following pre-requisites are applicable if the micro integration need to create LVQ if it doesn't exist.
+
+1. Ensure the Client Profile applied to the user has "Allow Client to Create Endpoints" enabled.
+2. Ensure the ACL applied to the user has publish and subscribe access to the topic subscribed by LVQ. The Subscribe access is required to programmatically apply the topic subscription to LVQ.
+3. The Micro Integration will create a LVQ with pre-requisites mentioned in above section.
+
 === User Authentication
 
 Solace Spark Connector supports Basic, Client Certificate and OAuth authentication to Solace. Client Credentials flow is supported when connecting using OAuth.

src/docs/sections/general/configuration/solace-spark-source-config.adoc

@@ -204,20 +204,6 @@ a| Set this value to true if connector needs to identify and acknowledge process
 
 NOTE: This property will be void if replay strategy is enabled.
 
-| offsetIndicator
-| string
-| any
-| MESSAGE_ID, CORRELATION_ID, APPLICATION_MESSAGE_ID, <CUSTOM_USER_PROPERTY>
-a| Set this value if your Solace Message has unique ID in message header. Supported Values are
-
- * MESSAGE_ID
- * CORRELATION_ID
- * APPLICATION_MESSAGE_ID
- * <CUSTOM_USER_PROPERTY> - refers to one of headers in user properties
-Header.
-
-Note: Default value uses replication group message ID property as offset indicator. ReplicationGroupMessageId is a unique message id across a broker cluster.
-
 | includeHeaders
 | boolean
 | true or false

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

@@ -192,6 +192,9 @@ private Optional<String> getExecutorLocation(List<String> executorLocations, int
     @Override
     public PartitionReaderFactory createReaderFactory() {
         log.info("SolaceSparkConnector - Create reader factory with includeHeaders :: {}", this.includeHeaders);
+        if(currentCheckpoint != null && currentCheckpoint.isEmpty()) {
+            currentCheckpoint = this.getCheckpoint();
+        }
         return new SolaceDataSourceReaderFactory(this.includeHeaders, this.properties, currentCheckpoint, this.checkpointLocation);
     }
 
@@ -217,6 +220,7 @@ public Offset deserializeOffset(String json) {
             log.info("SolaceSparkConnector - Deserialized offset {}", new Gson().toJson(solaceSourceOffset));
         }
         lastKnownOffsetId = solaceSourceOffset.getOffset();
+        currentCheckpoint = solaceSourceOffset.getCheckpoints();
         solaceSourceOffset.getCheckpoints().forEach(checkpoint -> lastKnownMessageIds = String.join(",", lastKnownMessageIds, checkpoint.getMessageIDs()));
 
         return solaceSourceOffset;
@@ -246,7 +250,7 @@ private SolaceSourceOffset getDeserializedOffset(String json) {
                 }
             } catch (Exception e2) {
                 log.error("SolaceSparkConnector - Exception when migrating offset to latest format.");
-                throw new RuntimeException("SolaceSparkConnector - Exception when migrating offset to latest format.", e);
+                throw new RuntimeException("SolaceSparkConnector - Exception when migrating offset to latest format. Please delete the checkpoint and restart the micro integration", e);
             }
         }
     }

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

@@ -35,7 +35,7 @@ public static void addMessage(String uniqueId, SolaceMessage message) {
         if(messages.containsKey(uniqueId)) {
             messageList = messages.get(uniqueId);
         }
-        messageList.add(message);
+        messageList.addIfAbsent(message);
         messages.put(uniqueId, messageList);
     }
 

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

@@ -67,6 +67,21 @@ public SolaceInputPartitionReader(SolaceInputPartition inputPartition, boolean i
 
         this.solaceInputPartition = inputPartition;
         this.uniqueId = this.solaceInputPartition.getId();
+        this.includeHeaders = includeHeaders;
+        this.properties = properties;
+        this.taskContext = taskContext;
+        this.taskId = taskContext.taskAttemptId();
+        this.checkpoints = checkpoints;
+        this.checkpointLocation = checkpointLocation;
+        this.batchSize = Integer.parseInt(properties.getOrDefault(SolaceSparkStreamingProperties.BATCH_SIZE, SolaceSparkStreamingProperties.BATCH_SIZE_DEFAULT));
+        this.receiveWaitTimeout = Long.parseLong(properties.getOrDefault(SolaceSparkStreamingProperties.QUEUE_RECEIVE_WAIT_TIMEOUT, SolaceSparkStreamingProperties.QUEUE_RECEIVE_WAIT_TIMEOUT_DEFAULT));
+        this.closeReceiversOnPartitionClose = Boolean.parseBoolean(properties.getOrDefault(SolaceSparkStreamingProperties.CLOSE_RECEIVERS_ON_PARTITION_CLOSE, SolaceSparkStreamingProperties.CLOSE_RECEIVERS_ON_PARTITION_CLOSE_DEFAULT));
+        boolean ackLastProcessedMessages = Boolean.parseBoolean(properties.getOrDefault(SolaceSparkStreamingProperties.ACK_LAST_PROCESSED_MESSAGES, SolaceSparkStreamingProperties.ACK_LAST_PROCESSED_MESSAGES_DEFAULT));
+        String replayStrategy = this.properties.getOrDefault(SolaceSparkStreamingProperties.REPLAY_STRATEGY, null);
+        if (replayStrategy != null && !replayStrategy.isEmpty()) {
+            ackLastProcessedMessages = false;
+        }
+
         String currentBatchId = taskContext.getLocalProperty(MicroBatchExecution.BATCH_ID_KEY());
         /*
          * In case when multiple operations are performed on dataframe, input partition will be called as part of Spark scan.
@@ -83,29 +98,26 @@ public SolaceInputPartitionReader(SolaceInputPartition inputPartition, boolean i
             SolaceMessageTracker.ackMessages(uniqueId);
             log.trace("SolaceSparkConnector - Total time taken to acknowledge messages {} ms", (System.currentTimeMillis() - startTime));
         } else {
+            log.info("SolaceSparkConnector - Spark Batch with id {} is requesting data again. It may be because of multiple operations on same dataframe.", currentBatchId);
             isCommitTriggered = false;
             CopyOnWriteArrayList<SolaceMessage> messageList = SolaceMessageTracker.getMessages(uniqueId);
             if (messageList != null) {
                 iterator = messageList.iterator();
+                if(messageList.size() < batchSize) {
+                    if(messageList.size() == 1) {
+                        log.info("SolaceSparkConnector - Only {} message is available from earlier request. This is most likely due to the result of isEmpty or similar operation on dataframe. Spark will immediately terminate input partition if one message is available from source.", messageList.size());
+                    }
+                    log.info("SolaceSparkConnector - Since only {} messages are available from earlier request. The remaining {} messages will be consumed from queue, if available, to fill the configured batch size of {}", messageList.size(), (batchSize - messageList.size()), this.batchSize);
+                } else {
+                    log.info("SolaceSparkConnector - {} messages available from the earlier request, matching the configured batch size of {}. The same messages will be returned as batch is not yet completed.", this.batchSize, messageList.size());
+                }
+
             }
         }
 
         SolaceMessageTracker.setLastBatchId(this.uniqueId, currentBatchId);
 
-        this.includeHeaders = includeHeaders;
-        this.properties = properties;
-        this.taskContext = taskContext;
-        this.taskId = taskContext.taskAttemptId();
-        this.checkpoints = checkpoints;
-        this.checkpointLocation = checkpointLocation;
-        this.batchSize = Integer.parseInt(properties.getOrDefault(SolaceSparkStreamingProperties.BATCH_SIZE, SolaceSparkStreamingProperties.BATCH_SIZE_DEFAULT));
-        this.receiveWaitTimeout = Long.parseLong(properties.getOrDefault(SolaceSparkStreamingProperties.QUEUE_RECEIVE_WAIT_TIMEOUT, SolaceSparkStreamingProperties.QUEUE_RECEIVE_WAIT_TIMEOUT_DEFAULT));
-        this.closeReceiversOnPartitionClose = Boolean.parseBoolean(properties.getOrDefault(SolaceSparkStreamingProperties.CLOSE_RECEIVERS_ON_PARTITION_CLOSE, SolaceSparkStreamingProperties.CLOSE_RECEIVERS_ON_PARTITION_CLOSE_DEFAULT));
-        boolean ackLastProcessedMessages = Boolean.parseBoolean(properties.getOrDefault(SolaceSparkStreamingProperties.ACK_LAST_PROCESSED_MESSAGES, SolaceSparkStreamingProperties.ACK_LAST_PROCESSED_MESSAGES_DEFAULT));
-        String replayStrategy = this.properties.getOrDefault(SolaceSparkStreamingProperties.REPLAY_STRATEGY, null);
-        if (replayStrategy == null || replayStrategy.isEmpty()) {
-            ackLastProcessedMessages = false;
-        }
+
 
         log.info("SolaceSparkConnector - Checking for connection {}", inputPartition.getId());
 
@@ -131,7 +143,7 @@ public SolaceInputPartitionReader(SolaceInputPartition inputPartition, boolean i
             createNewConnection(inputPartition.getId(), ackLastProcessedMessages);
         }
         checkException();
-        log.info("SolaceSparkConnector - Acquired connection to Solace broker for partition {}", inputPartition.getId());
+        log.info("SolaceSparkConnector - Current consumer session on input partition {} is {}", inputPartition.getId(), this.solaceBroker.getUniqueName());
         registerTaskListener();
     }
 
@@ -209,7 +221,7 @@ private SolaceMessage getNextMessage() {
           If commit is triggered or messageList is null we need to fetch messages from Solace.
           In case of same batch just return the available messages in message tracker.
          */
-        if (this.isCommitTriggered || iterator == null || !iterator.hasNext()) {
+        if (this.isCommitTriggered || iterator == null) {
             LinkedBlockingQueue<SolaceMessage> queue = solaceBroker.getMessages(0);
             if (queue != null) {
                 while (shouldProcessMoreMessages(batchSize, messages)) {
@@ -254,6 +266,29 @@ private SolaceMessage getNextMessage() {
                             return solaceMessage;
                         }
                     } else {
+                        LinkedBlockingQueue<SolaceMessage> queue = solaceBroker.getMessages(0);
+                        if (queue != null) {
+                            try {
+                                solaceMessage = queue.poll(receiveWaitTimeout, TimeUnit.MILLISECONDS);
+                                if (solaceMessage == null) {
+                                    return null;
+                                }
+
+                                if (batchSize > 0) {
+                                    messages++;
+                                }
+                                if (isMessageAlreadyProcessed(solaceMessage)) {
+                                    log.info("Message is added to previous partitions for processing. Moving to next message");
+                                } else {
+                                    return solaceMessage;
+                                }
+
+                            } catch (InterruptedException | SDTException e) {
+                                log.warn("No messages available within specified receiveWaitTimeout", e);
+                                Thread.currentThread().interrupt();
+                                return null;
+                            }
+                        }
                         return null;
                     }
                 } catch (Exception e) {
@@ -303,39 +338,40 @@ private void registerTaskListener() {
             if (context.isInterrupted() || context.isFailed()) {
                 logShutdownMessage(context);
             } else if (context.isCompleted()) {
-                List<String> ids = SolaceMessageTracker.getIds();
-                try {
-                    Path path = Paths.get(this.checkpointLocation + "/" + this.solaceInputPartition.getId() + ".txt");
-                    log.trace("SolaceSparkConnector - File path {} to store checkpoint processed in worker node {}", path.toString(), this.solaceInputPartition.getPreferredLocation());
-                    Path parentDir = path.getParent();
-                    if (parentDir != null) {
-                        // Create the directory and all nonexistent parent directories
-                        Files.createDirectories(parentDir);
-                        log.trace("SolaceSparkConnector - Created parent directory {} for file path {}", parentDir.toString(), path.toString());
-                    }
-                    // overwrite checkpoint to preserve latest value
-                    try (BufferedWriter writer = Files.newBufferedWriter(path, StandardOpenOption.CREATE,
-                            StandardOpenOption.TRUNCATE_EXISTING)) {
-                        for (String id : ids) {
-                            String processedMessageIDs = SolaceMessageTracker.getProcessedMessagesIDs(id);
-                            if (processedMessageIDs != null) {
-                                SolaceSparkPartitionCheckpoint solaceSparkPartitionCheckpoint = new SolaceSparkPartitionCheckpoint(processedMessageIDs, id);
-                                CopyOnWriteArrayList<SolaceSparkPartitionCheckpoint> solaceSparkPartitionCheckpoints = new CopyOnWriteArrayList<>();
-                                solaceSparkPartitionCheckpoints.add(solaceSparkPartitionCheckpoint);
-                                // Publish state to checkpoint. On commit the state is published to Solace LVQ.
-                                writer.write(new Gson().toJson(solaceSparkPartitionCheckpoints));
-                                writer.newLine();
-                                log.trace("SolaceSparkConnector - Checkpoint {} stored in file path {}", new Gson().toJson(solaceSparkPartitionCheckpoints), path.toString());
-                                SolaceMessageTracker.removeProcessedMessagesIDs(id);
-                            }
+                String processedMessageIDs = SolaceMessageTracker.getProcessedMessagesIDs(this.solaceInputPartition.getId());
+                System.out.println(this.solaceInputPartition.getId() + " - " + processedMessageIDs);
+                if(processedMessageIDs != null && !processedMessageIDs.isEmpty()) {
+                    try {
+                        Path path = Paths.get(this.checkpointLocation + "/" + this.solaceInputPartition.getId() + ".txt");
+                        log.trace("SolaceSparkConnector - File path {} to store checkpoint processed in worker node {}", path.toString(), this.solaceInputPartition.getPreferredLocation());
+                        Path parentDir = path.getParent();
+                        if (parentDir != null) {
+                            // Create the directory and all nonexistent parent directories
+                            Files.createDirectories(parentDir);
+                            log.trace("SolaceSparkConnector - Created parent directory {} for file path {}", parentDir.toString(), path.toString());
                         }
+                        // overwrite checkpoint to preserve latest value
+                        try (BufferedWriter writer = Files.newBufferedWriter(path, StandardOpenOption.CREATE,
+                                StandardOpenOption.TRUNCATE_EXISTING)) {
+//                        for (String id : ids) {
+                            SolaceSparkPartitionCheckpoint solaceSparkPartitionCheckpoint = new SolaceSparkPartitionCheckpoint(processedMessageIDs, this.solaceInputPartition.getId());
+                            CopyOnWriteArrayList<SolaceSparkPartitionCheckpoint> solaceSparkPartitionCheckpoints = new CopyOnWriteArrayList<>();
+                            solaceSparkPartitionCheckpoints.add(solaceSparkPartitionCheckpoint);
+                            // Publish state to checkpoint. On commit the state is published to Solace LVQ.
+                            writer.write(new Gson().toJson(solaceSparkPartitionCheckpoints));
+                            writer.newLine();
+                            log.trace("SolaceSparkConnector - Checkpoint {} stored in file path {}", new Gson().toJson(solaceSparkPartitionCheckpoints), path.toString());
+                            SolaceMessageTracker.removeProcessedMessagesIDs(this.solaceInputPartition.getId());
+                            //                        }
+                        }
+                    } catch (IOException e) {
+                        log.error("SolaceSparkConnector - Exception when writing checkpoint to path {}", this.checkpointLocation, e);
+                        this.solaceBroker.close();
+                        throw new RuntimeException(e);
                     }
-                } catch (IOException e) {
-                    log.error("SolaceSparkConnector - Exception when writing checkpoint to path {}", this.checkpointLocation, e);
-                    this.solaceBroker.close();
-                    throw new RuntimeException(e);
                 }
 
+
                 log.info("SolaceSparkConnector - Total time taken by executor is {} ms for Task {}", context.taskMetrics().executorRunTime(), uniqueId);
 
                 if (closeReceiversOnPartitionClose) {