DATAGO-92181: Solace message replay support (#23)

Author: SravanThotakura05

pubsubplus-connector-spark_3.x/src/docs/asciidoc/User-Guide.adoc

@@ -85,6 +85,10 @@ If rotating access token is present in file accessible by connector use below op
 
 NOTE: When access token is read from file, it may lose some of it's expiry time by the time it is accessed by connector. It is recommended to have minimal time difference between writing to file and access by the connector so that a valid new token is updated in solace session before expiry of old token.
 
+=== Message Replay
+
+Solace Spark Connector can replay messages using Solace Replay Log. Connector can replay all messages or after specific replication group message id or after specific timestamp. Please refer to https://docs.solace.com/Features/Replay/Msg-Replay-Concepts-Config.htm[Message Replay Configuration] to enable replay log in Solace PubSub+ broker.
+
 === Solace Spark Streaming Schema Structure
 
 Solace Spark Connector transforms the incoming message to Spark row with below schema definition.

pubsubplus-connector-spark_3.x/src/docs/sections/general/configuration/spark-config.adoc

@@ -55,7 +55,7 @@
 | int
 | 0 - 60000
 | 3000
-| How much time in (MS) to wait between each attempt to connect or reconnect to a host. If a connect or reconnect attempt to host is not successful, the API waits for the amount of time set for reconnectRetryWaitInMillis, and then makes another connect or reconnect attempt.
+| How much time in (MS) to wait between each attempt to connect or reconnect to a host. If connect or reconnect attempt to host is not successful, the API waits for the amount of time set for reconnectRetryWaitInMillis, and then makes another connect or reconnect attempt.
 
 | solace.apiProperties.<Property>
 | any
@@ -154,17 +154,37 @@ solace.apiProperties.client_channel_properties.keepAliveIntervalInMillis=3000
 | 1
 | Set number of messages to be processed in batch. The connector can stream data in batches to Spark based on configured size.
 
+| replayStrategy
+| string
+| BEGINNING \| TIMEBASED \| REPLICATION-GROUP-MESSAGE-ID
+| empty
+| Set the replay strategy if messages need to be replayed from broker to connector. For more information refer to https://docs.solace.com/Features/Replay/Msg-Replay-Concepts-Config.htm#Types[SolaceReplayConfiguration]
+
+| replayStrategyReplicationGroupMessageId
+| string
+| valid-replication-group-message-id
+| empty
+| Set the property if replay strategy is REPLICATION-GROUP-MESSAGE-ID. Message playback is started after this replication group message id.
+
+| replayStrategyStartTime
+| string
+| datetime string<yyyy-MM-dd'T'HH:mm:ss>
+| empty
+| Set the property if replay strategy is TIMEBASED. Any messages in the replay log equal to, or newer than, the specified date and time that match the endpoint’s subscriptions are replayed to the connector. The date can't be earlier than the date the replay log was created, otherwise replay will fail.
+
+| replayStrategyTimeZone
+| string
+| valid timezone
+| UTC
+| Set the property if replay strategy is TIMEBASED.
+
 | ackLastProcessedMessages
 | boolean
 | true or false
 | false
-| Set this value to true if connector needs to identify and acknowledge processed messages in last run during restarts. The connector purely depends on checkpoint generated during Spark commit. We recommended enabling this configuration only when your downstream system has processed data in previous run.
+a| Set this value to true if connector needs to identify and acknowledge processed messages in last run during restarts. The connector purely depends on checkpoint generated during Spark commit. We recommended enabling this configuration only when your downstream system has processed data in previous run.
 
-| skipDuplicates
-| boolean
-| true or false
-| false
-| Set this value to true if connector needs check for duplicates before adding to Spark row. This scenario occurs when a task is running late and new task is started. The new task may have duplicate message as messages from earlier are not acknowledged by the time it is start.
+NOTE: This property will be void if replay strategy is enabled.
 
 | offsetIndicator
 | string

pubsubplus-connector-spark_3.x/src/main/java/com/solacecoe/connectors/spark/streaming/SolaceMicroBatch.java

@@ -40,9 +40,9 @@ public class SolaceMicroBatch implements MicroBatchStream, SupportsAdmissionCont
     private final int batchSize;
     private final int partitions;
     private final boolean ackLastProcessedMessages;
-    private final boolean skipMessageReprocessingIfTasksAreRunningLate;
     private final boolean createFlowsOnSameSession;
     private final boolean includeHeaders;
+    private final boolean initiateReplay;
 
     public SolaceMicroBatch(StructType schema, Map<String, String> properties, CaseInsensitiveStringMap options) {
         log.info("SolaceSparkConnector - Initializing Solace Spark Connector");
@@ -112,7 +112,6 @@ public SolaceMicroBatch(StructType schema, Map<String, String> properties, CaseI
         }
 
         ackLastProcessedMessages = Boolean.parseBoolean(properties.getOrDefault(SolaceSparkStreamingProperties.ACK_LAST_PROCESSED_MESSAGES, SolaceSparkStreamingProperties.ACK_LAST_PROCESSED_MESSAGES_DEFAULT));
-        skipMessageReprocessingIfTasksAreRunningLate = Boolean.parseBoolean(properties.getOrDefault(SolaceSparkStreamingProperties.SKIP_DUPLICATES, SolaceSparkStreamingProperties.SKIP_DUPLICATES_DEFAULT));
         log.info("SolaceSparkConnector - Ack Last processed messages is set to {}", ackLastProcessedMessages);
 
         includeHeaders = Boolean.parseBoolean(properties.getOrDefault(SolaceSparkStreamingProperties.INCLUDE_HEADERS, SolaceSparkStreamingProperties.INCLUDE_HEADERS_DEFAULT));
@@ -133,11 +132,12 @@ public SolaceMicroBatch(StructType schema, Map<String, String> properties, CaseI
 
         solaceConnectionManager = new SolaceConnectionManager();
         log.info("SolaceSparkConnector - Solace Connection Details Host : {}, VPN : {}, Username : {}", properties.get(SolaceSparkStreamingProperties.HOST), properties.get(SolaceSparkStreamingProperties.VPN), properties.get(SolaceSparkStreamingProperties.USERNAME));
-        SolaceBroker solaceBroker = new SolaceBroker(properties.get(SolaceSparkStreamingProperties.HOST), properties.get(SolaceSparkStreamingProperties.VPN), properties.get(SolaceSparkStreamingProperties.USERNAME), properties.get(SolaceSparkStreamingProperties.PASSWORD), properties.get(SolaceSparkStreamingProperties.QUEUE), properties);
+        this.initiateReplay = properties.getOrDefault(SolaceSparkStreamingProperties.REPLAY_STRATEGY ,null) != null;
+        SolaceBroker solaceBroker = new SolaceBroker(properties);
         solaceConnectionManager.addConnection(solaceBroker);
         for (int i = 0; i < partitions; i++) {
             if(!createFlowsOnSameSession && i > 0) {
-                solaceBroker = new SolaceBroker(properties.get(SolaceSparkStreamingProperties.HOST), properties.get(SolaceSparkStreamingProperties.VPN), properties.get(SolaceSparkStreamingProperties.USERNAME), properties.get(SolaceSparkStreamingProperties.PASSWORD), properties.get(SolaceSparkStreamingProperties.QUEUE), properties);
+                solaceBroker = new SolaceBroker(properties);
                 solaceConnectionManager.addConnection(solaceBroker);
             }
             EventListener eventListener = new EventListener((i + 1));
@@ -198,7 +198,7 @@ private InputPartition[] splitDataOnPartitions() {
                                 this.messages.put(solaceRecord.getMessageId(), solaceMessage);
                             } else {
                                 this.messages.put(solaceRecord.getMessageId(), solaceMessage);
-                                if (ackLastProcessedMessages) {
+                                if (ackLastProcessedMessages && !this.initiateReplay) {
                                     log.info("SolaceSparkConnector - Ack last processed messages is enabled. Checking if message is already processed based on available offsets.");
                                     // based on last successful offset, extract the message ID and see if same message is received, if so ack the message
                                     if (offsetJson != null && offsetJson.has("messageIDs")) {

pubsubplus-connector-spark_3.x/src/main/java/com/solacecoe/connectors/spark/streaming/properties/SolaceSparkStreamingProperties.java

@@ -8,6 +8,10 @@ public class SolaceSparkStreamingProperties {
     public static final String QUEUE = "queue";
     public static final String BATCH_SIZE = "batchSize";
     public static final String BATCH_SIZE_DEFAULT = "1";
+    public static final String REPLAY_STRATEGY = "replayStrategy";
+    public static final String REPLAY_STRATEGY_REPLICATION_GROUP_MESSAGE_ID = "replayStrategyReplicationGroupMessageId";
+    public static final String REPLAY_STRATEGY_START_TIME = "replayStrategyStartTime";
+    public static final String REPLAY_STRATEGY_TIMEZONE = "replayStrategyTimeZone";
     public static final String ACK_LAST_PROCESSED_MESSAGES = "ackLastProcessedMessages";
     public static final String ACK_LAST_PROCESSED_MESSAGES_DEFAULT = "false";
     public static final String SKIP_DUPLICATES = "skipDuplicates";