DATAGO-97452: Check username & password only if authentication is basic

pom.xml

@@ -10,7 +10,7 @@
     <packaging>pom</packaging>
     <name>PubSubPlus Connector Spark - Parent</name>
     <description>Solace PubSub+ Connector for Spark streams data from Solace PubSub+ broker to Spark Data Sources.</description>
-    <url>https://solace.com</url>
+    <url>https://solace.com/integration-hub/apache-spark/</url>
 
     <modules>
         <module>pubsubplus-connector-spark_3.x</module>

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

@@ -9,7 +9,7 @@
 import com.solacecoe.connectors.spark.streaming.solace.LVQEventListener;
 import com.solacecoe.connectors.spark.streaming.solace.SolaceBroker;
 import com.solacecoe.connectors.spark.streaming.solace.exceptions.SolaceInvalidPropertyException;
-import com.solacesystems.jcsmp.JCSMPProperties;
+import com.solacecoe.connectors.spark.streaming.solace.utils.SolaceUtils;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.apache.spark.SparkEnv;
@@ -48,48 +48,7 @@ public SolaceMicroBatch(Map<String, String> properties) {
         log.info("SolaceSparkConnector - Initializing Solace Spark Connector");
         // Initialize classes required for Solace connectivity
 
-        // User configuration validation
-        if(!properties.containsKey(SolaceSparkStreamingProperties.HOST) || properties.get(SolaceSparkStreamingProperties.HOST) == null || properties.get(SolaceSparkStreamingProperties.HOST).isEmpty()) {
-            throw new SolaceInvalidPropertyException("SolaceSparkConnector - Please provide Solace Host name in configuration options");
-        }
-        if(!properties.containsKey(SolaceSparkStreamingProperties.VPN) || properties.get(SolaceSparkStreamingProperties.VPN) == null || properties.get(SolaceSparkStreamingProperties.VPN).isEmpty()) {
-            throw new SolaceInvalidPropertyException("SolaceSparkConnector - Please provide Solace VPN name in configuration options");
-        }
-
-        if(properties.containsKey(SolaceSparkStreamingProperties.SOLACE_API_PROPERTIES_PREFIX+ JCSMPProperties.AUTHENTICATION_SCHEME) &&
-                properties.get(SolaceSparkStreamingProperties.SOLACE_API_PROPERTIES_PREFIX+ JCSMPProperties.AUTHENTICATION_SCHEME).equals(JCSMPProperties.AUTHENTICATION_SCHEME_OAUTH2)) {
-            if(!properties.containsKey(SolaceSparkStreamingProperties.OAUTH_CLIENT_ACCESSTOKEN)) {
-                if(!properties.containsKey(SolaceSparkStreamingProperties.OAUTH_CLIENT_AUTHSERVER_URL) || properties.get(SolaceSparkStreamingProperties.OAUTH_CLIENT_AUTHSERVER_URL) == null || properties.get(SolaceSparkStreamingProperties.OAUTH_CLIENT_AUTHSERVER_URL).isEmpty()) {
-                    throw new SolaceInvalidPropertyException("SolaceSparkConnector - Please provide OAuth Client Authentication Server URL");
-                }
-
-                if(!properties.containsKey(SolaceSparkStreamingProperties.OAUTH_CLIENT_CLIENT_ID) || properties.get(SolaceSparkStreamingProperties.OAUTH_CLIENT_CLIENT_ID) == null || properties.get(SolaceSparkStreamingProperties.OAUTH_CLIENT_CLIENT_ID).isEmpty()) {
-                    throw new SolaceInvalidPropertyException("SolaceSparkConnector - Please provide OAuth Client ID");
-                }
-
-                if(!properties.containsKey(SolaceSparkStreamingProperties.OAUTH_CLIENT_CREDENTIALS_CLIENTSECRET) || properties.get(SolaceSparkStreamingProperties.OAUTH_CLIENT_CREDENTIALS_CLIENTSECRET) == null || properties.get(SolaceSparkStreamingProperties.OAUTH_CLIENT_CREDENTIALS_CLIENTSECRET).isEmpty()) {
-                    throw new SolaceInvalidPropertyException("SolaceSparkConnector - Please provide OAuth Client Credentials Secret");
-                }
-
-                String clientCertificate = properties.getOrDefault(SolaceSparkStreamingProperties.OAUTH_CLIENT_AUTHSERVER_CLIENT_CERTIFICATE, null);
-                if(clientCertificate != null) {
-                    String trustStoreFilePassword = properties.getOrDefault(SolaceSparkStreamingProperties.OAUTH_CLIENT_AUTHSERVER_TRUSTSTORE_PASSWORD, null);
-                    if (trustStoreFilePassword == null || trustStoreFilePassword.isEmpty()) {
-                        throw new SolaceInvalidPropertyException("SolaceSparkConnector - Please provide OAuth Client TrustStore Password. If TrustStore file path is not configured, please provide password for default java truststore");
-                    }
-                }
-            } else if(properties.getOrDefault(SolaceSparkStreamingProperties.OAUTH_CLIENT_ACCESSTOKEN, null) == null || properties.getOrDefault(SolaceSparkStreamingProperties.OAUTH_CLIENT_ACCESSTOKEN, null).isEmpty()) {
-                throw new SolaceInvalidPropertyException("SolaceSparkConnector - Please provide valid access token input");
-            }
-        } else {
-            if (!properties.containsKey(SolaceSparkStreamingProperties.USERNAME) || properties.get(SolaceSparkStreamingProperties.USERNAME) == null || properties.get(SolaceSparkStreamingProperties.USERNAME).isEmpty()) {
-                throw new SolaceInvalidPropertyException("SolaceSparkConnector - Please provide Solace Username in configuration options");
-            }
-
-            if (!properties.containsKey(SolaceSparkStreamingProperties.PASSWORD) || properties.get(SolaceSparkStreamingProperties.PASSWORD) == null || properties.get(SolaceSparkStreamingProperties.PASSWORD).isEmpty()) {
-                throw new SolaceInvalidPropertyException("SolaceSparkConnector - Please provide Solace Password in configuration options");
-            }
-        }
+        SolaceUtils.validateCommonProperties(properties);
 
         if(!properties.containsKey(SolaceSparkStreamingProperties.QUEUE) || properties.get(SolaceSparkStreamingProperties.QUEUE) == null || properties.get(SolaceSparkStreamingProperties.QUEUE).isEmpty()) {
             throw new SolaceInvalidPropertyException("SolaceSparkConnector - Please provide Solace Queue in configuration options");

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

@@ -31,11 +31,15 @@ public SolaceDataSourceReaderFactory(boolean includeHeaders, String lastKnownOff
 
     @Override
     public PartitionReader<InternalRow> createReader(InputPartition partition) {
-        TaskContext taskCtx = TaskContext.get();
-        String queryId = taskCtx.getLocalProperty(StreamExecution.QUERY_ID_KEY());
-        String batchId = taskCtx.getLocalProperty(MicroBatchExecution.BATCH_ID_KEY());
-        SolaceInputPartition solaceInputPartition = (SolaceInputPartition) partition;
-        log.info("SolaceSparkConnector - Creating reader for input partition reader factory with query id {}, batch id {}, task id {} and partition id {}", queryId, batchId, taskCtx.taskAttemptId(), taskCtx.partitionId());
-        return new SolaceInputPartitionReader(solaceInputPartition, includeHeaders, lastKnownOffset, properties, taskCtx, checkpoints);
+        try {
+            TaskContext taskCtx = TaskContext.get();
+            String queryId = taskCtx.getLocalProperty(StreamExecution.QUERY_ID_KEY());
+            String batchId = taskCtx.getLocalProperty(MicroBatchExecution.BATCH_ID_KEY());
+            SolaceInputPartition solaceInputPartition = (SolaceInputPartition) partition;
+            log.info("SolaceSparkConnector - Creating reader for input partition reader factory with query id {}, batch id {}, task id {} and partition id {}", queryId, batchId, taskCtx.taskAttemptId(), taskCtx.partitionId());
+            return new SolaceInputPartitionReader(solaceInputPartition, includeHeaders, lastKnownOffset, properties, taskCtx, checkpoints);
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
     }
 }

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

@@ -93,6 +93,9 @@ public SolaceInputPartitionReader(SolaceInputPartition inputPartition, boolean i
 
     @Override
     public boolean next() {
+        if(this.solaceBroker != null && this.solaceBroker.isException()) {
+            throw new SolaceSessionException(this.solaceBroker.getException());
+        }
         solaceMessage = getNextMessage();
         return solaceMessage != null;
     }

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

@@ -91,8 +91,8 @@ public SolaceBroker(Map<String, String> properties, String clientType) {
                     scheduleOAuthRefresh(interval);
                 }
             } else {
-                jcsmpProperties.setProperty(JCSMPProperties.USERNAME, properties.get(SolaceSparkStreamingProperties.USERNAME)); // client-username
-                jcsmpProperties.setProperty(JCSMPProperties.PASSWORD, properties.get(SolaceSparkStreamingProperties.PASSWORD)); // client-password
+                jcsmpProperties.setProperty(JCSMPProperties.USERNAME, properties.getOrDefault(SolaceSparkStreamingProperties.USERNAME, "")); // client-username
+                jcsmpProperties.setProperty(JCSMPProperties.PASSWORD, properties.getOrDefault(SolaceSparkStreamingProperties.PASSWORD, "")); // client-password
             }
 
             this.uniqueName = JCSMPFactory.onlyInstance().createUniqueName("solace/spark/connector/"+clientType);

pubsubplus-connector-spark_3.x/src/main/java/com/solacecoe/connectors/spark/streaming/solace/utils/SolaceUtils.java

@@ -2,6 +2,7 @@
 
 import com.solacecoe.connectors.spark.streaming.properties.SolaceHeaderMeta;
 import com.solacecoe.connectors.spark.streaming.properties.SolaceSparkStreamingProperties;
+import com.solacecoe.connectors.spark.streaming.solace.exceptions.SolaceInvalidPropertyException;
 import com.solacesystems.jcsmp.*;
 
 import java.io.Serializable;
@@ -13,6 +14,51 @@
 
 public class SolaceUtils {
 
+    public static void validateCommonProperties(Map<String, String> properties) {
+        // User configuration validation
+        if(!properties.containsKey(SolaceSparkStreamingProperties.HOST) || properties.get(SolaceSparkStreamingProperties.HOST) == null || properties.get(SolaceSparkStreamingProperties.HOST).isEmpty()) {
+            throw new SolaceInvalidPropertyException("SolaceSparkConnector - Please provide Solace Host name in configuration options");
+        }
+        if(!properties.containsKey(SolaceSparkStreamingProperties.VPN) || properties.get(SolaceSparkStreamingProperties.VPN) == null || properties.get(SolaceSparkStreamingProperties.VPN).isEmpty()) {
+            throw new SolaceInvalidPropertyException("SolaceSparkConnector - Please provide Solace VPN name in configuration options");
+        }
+
+        if(properties.containsKey(SolaceSparkStreamingProperties.SOLACE_API_PROPERTIES_PREFIX+ JCSMPProperties.AUTHENTICATION_SCHEME) &&
+                properties.get(SolaceSparkStreamingProperties.SOLACE_API_PROPERTIES_PREFIX+ JCSMPProperties.AUTHENTICATION_SCHEME).equals(JCSMPProperties.AUTHENTICATION_SCHEME_OAUTH2)) {
+            if(!properties.containsKey(SolaceSparkStreamingProperties.OAUTH_CLIENT_ACCESSTOKEN)) {
+                if(!properties.containsKey(SolaceSparkStreamingProperties.OAUTH_CLIENT_AUTHSERVER_URL) || properties.get(SolaceSparkStreamingProperties.OAUTH_CLIENT_AUTHSERVER_URL) == null || properties.get(SolaceSparkStreamingProperties.OAUTH_CLIENT_AUTHSERVER_URL).isEmpty()) {
+                    throw new SolaceInvalidPropertyException("SolaceSparkConnector - Please provide OAuth Client Authentication Server URL");
+                }
+
+                if(!properties.containsKey(SolaceSparkStreamingProperties.OAUTH_CLIENT_CLIENT_ID) || properties.get(SolaceSparkStreamingProperties.OAUTH_CLIENT_CLIENT_ID) == null || properties.get(SolaceSparkStreamingProperties.OAUTH_CLIENT_CLIENT_ID).isEmpty()) {
+                    throw new SolaceInvalidPropertyException("SolaceSparkConnector - Please provide OAuth Client ID");
+                }
+
+                if(!properties.containsKey(SolaceSparkStreamingProperties.OAUTH_CLIENT_CREDENTIALS_CLIENTSECRET) || properties.get(SolaceSparkStreamingProperties.OAUTH_CLIENT_CREDENTIALS_CLIENTSECRET) == null || properties.get(SolaceSparkStreamingProperties.OAUTH_CLIENT_CREDENTIALS_CLIENTSECRET).isEmpty()) {
+                    throw new SolaceInvalidPropertyException("SolaceSparkConnector - Please provide OAuth Client Credentials Secret");
+                }
+
+                String clientCertificate = properties.getOrDefault(SolaceSparkStreamingProperties.OAUTH_CLIENT_AUTHSERVER_CLIENT_CERTIFICATE, null);
+                if(clientCertificate != null) {
+                    String trustStoreFilePassword = properties.getOrDefault(SolaceSparkStreamingProperties.OAUTH_CLIENT_AUTHSERVER_TRUSTSTORE_PASSWORD, null);
+                    if (trustStoreFilePassword == null || trustStoreFilePassword.isEmpty()) {
+                        throw new SolaceInvalidPropertyException("SolaceSparkConnector - Please provide OAuth Client TrustStore Password. If TrustStore file path is not configured, please provide password for default java truststore");
+                    }
+                }
+            } else if(properties.getOrDefault(SolaceSparkStreamingProperties.OAUTH_CLIENT_ACCESSTOKEN, null) == null || properties.getOrDefault(SolaceSparkStreamingProperties.OAUTH_CLIENT_ACCESSTOKEN, null).isEmpty()) {
+                throw new SolaceInvalidPropertyException("SolaceSparkConnector - Please provide valid access token input");
+            }
+        } else if(properties.containsKey(SolaceSparkStreamingProperties.SOLACE_API_PROPERTIES_PREFIX+ JCSMPProperties.AUTHENTICATION_SCHEME) &&
+                properties.getOrDefault(SolaceSparkStreamingProperties.SOLACE_API_PROPERTIES_PREFIX+ JCSMPProperties.AUTHENTICATION_SCHEME, JCSMPProperties.AUTHENTICATION_SCHEME_BASIC).equals(JCSMPProperties.AUTHENTICATION_SCHEME_BASIC)){
+            if (!properties.containsKey(SolaceSparkStreamingProperties.USERNAME) || properties.get(SolaceSparkStreamingProperties.USERNAME) == null || properties.get(SolaceSparkStreamingProperties.USERNAME).isEmpty()) {
+                throw new SolaceInvalidPropertyException("SolaceSparkConnector - Please provide Solace Username in configuration options");
+            }
+
+            if (!properties.containsKey(SolaceSparkStreamingProperties.PASSWORD) || properties.get(SolaceSparkStreamingProperties.PASSWORD) == null || properties.get(SolaceSparkStreamingProperties.PASSWORD).isEmpty()) {
+                throw new SolaceInvalidPropertyException("SolaceSparkConnector - Please provide Solace Password in configuration options");
+            }
+        }
+    }
     public static String getMessageID(BytesXMLMessage message, String solaceOffsetIndicator) throws SDTException {
         switch (solaceOffsetIndicator) {
             case "CORRELATION_ID":

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

@@ -1,5 +1,6 @@
 package com.solacecoe.connectors.spark.streaming.write;
 
+import com.solacecoe.connectors.spark.streaming.solace.utils.SolaceUtils;
 import org.apache.spark.sql.connector.write.PhysicalWriteInfo;
 import org.apache.spark.sql.connector.write.WriterCommitMessage;
 import org.apache.spark.sql.connector.write.streaming.StreamingDataWriterFactory;
@@ -22,24 +23,7 @@ public SolaceStreamingWrite(StructType schema, Map<String, String> properties, C
         this.properties = properties;
         this.options = options;
 
-        if(!properties.containsKey("host") || properties.get("host") == null || properties.get("host").isEmpty()) {
-            log.error("SolaceSparkConnector - Please provide Solace Host name in configuration options");
-            throw new RuntimeException("SolaceSparkConnector - Please provide Solace Host name in configuration options");
-        }
-        if(!properties.containsKey("vpn") || properties.get("vpn") == null || properties.get("vpn").isEmpty()) {
-            log.error("SolaceSparkConnector - Please provide Solace VPN name in configuration options");
-            throw new RuntimeException("SolaceSparkConnector - Please provide Solace VPN name in configuration options");
-        }
-
-        if(!properties.containsKey("username") || properties.get("username") == null || properties.get("username").isEmpty()) {
-            log.error("SolaceSparkConnector - Please provide Solace Username in configuration options");
-            throw new RuntimeException("SolaceSparkConnector - Please provide Solace Username in configuration options");
-        }
-
-        if(!properties.containsKey("password") || properties.get("password") == null || properties.get("password").isEmpty()) {
-            log.error("SolaceSparkConnector - Please provide Solace Password in configuration options");
-            throw new RuntimeException("SolaceSparkConnector - Please provide Solace Password in configuration options");
-        }
+        SolaceUtils.validateCommonProperties(properties);
 
 //        if(!properties.containsKey("topic") || properties.get("topic") == null || properties.get("topic").isEmpty()) {
 //            log.error("SolaceSparkConnector - Please provide Solace Queue name in configuration options");