More StringUtils sanity checking

Most/all of these looked safe, but switching to StringUtils for consistency and to ensure that an empty string can't cause an issue.

Author: rjrudin

src/main/java/com/marklogic/kafka/connect/DefaultDatabaseClientConfigBuilder.java

@@ -7,6 +7,7 @@
 import com.marklogic.client.ext.helper.LoggingObject;
 import com.marklogic.client.ext.modulesloader.ssl.SimpleX509TrustManager;
 import org.apache.kafka.common.config.types.Password;
+import org.springframework.util.StringUtils;
 
 import javax.net.ssl.*;
 import java.io.InputStream;
@@ -28,11 +29,11 @@ public DatabaseClientConfig buildDatabaseClientConfig(Map<String, Object> parsed
         clientConfig.setSecurityContextType(SecurityContextType.valueOf(securityContextType));
 
         String database = (String) parsedConfig.get(MarkLogicConfig.CONNECTION_DATABASE);
-        if (database != null && database.trim().length() > 0) {
+        if (StringUtils.hasText(database)) {
             clientConfig.setDatabase(database);
         }
         String connType = (String) parsedConfig.get(MarkLogicConfig.CONNECTION_TYPE);
-        if (connType != null && connType.trim().length() > 0) {
+        if (StringUtils.hasText(connType)) {
             clientConfig.setConnectionType(DatabaseClient.ConnectionType.valueOf(connType.toUpperCase()));
         }
 

src/main/java/com/marklogic/kafka/connect/sink/WriteBatcherSinkTask.java

@@ -53,7 +53,7 @@ protected void onStart(Map<String, Object> parsedConfig) {
         configureWriteBatcher(parsedConfig, writeBatcher);
 
         final String flowName = (String) parsedConfig.get(MarkLogicSinkConfig.DATAHUB_FLOW_NAME);
-        if (flowName != null && flowName.trim().length() > 0) {
+        if (StringUtils.hasText(flowName)) {
             writeBatcher.onBatchSuccess(buildRunFlowListener(flowName, parsedConfig, databaseClientConfig));
         }
 
@@ -180,7 +180,7 @@ protected RunFlowWriteBatchListener buildRunFlowListener(String flowName, Map<St
         String logMessage = String.format("After ingesting a batch, will run flow '%s'", flowName);
         final String flowSteps = (String) parsedConfig.get(MarkLogicSinkConfig.DATAHUB_FLOW_STEPS);
         List<String> steps = null;
-        if (flowSteps != null && flowSteps.trim().length() > 0) {
+        if (StringUtils.hasText(flowSteps)) {
             steps = Arrays.asList(flowSteps.split(","));
             logMessage += String.format(" with steps '%s' constrained to the URIs in that batch", steps);
         }
@@ -205,9 +205,9 @@ protected Optional<ServerTransform> buildServerTransform(final Map<String, Objec
         if (StringUtils.hasText(transformName)) {
             ServerTransform transform = new ServerTransform(transformName);
             String params = (String) parsedConfig.get(MarkLogicSinkConfig.DMSDK_TRANSFORM_PARAMS);
-            if (params != null && params.trim().length() > 0) {
+            if (StringUtils.hasText(params)) {
                 String delimiter = (String) parsedConfig.get(MarkLogicSinkConfig.DMSDK_TRANSFORM_PARAMS_DELIMITER);
-                if (delimiter != null && delimiter.trim().length() > 0) {
+                if (StringUtils.hasText(delimiter)) {
                     addTransformParameters(transform, params, delimiter);
                 } else {
                     logger.warn("Unable to apply transform parameters to transform: {}; please set the " +

src/main/java/com/marklogic/kafka/connect/source/AbstractPlanInvoker.java

@@ -4,6 +4,7 @@
 import com.marklogic.client.ext.helper.LoggingObject;
 import com.marklogic.client.row.RowManager;
 import com.marklogic.kafka.connect.ConfigUtil;
+import org.springframework.util.StringUtils;
 
 import java.util.Map;
 
@@ -19,7 +20,7 @@ abstract class AbstractPlanInvoker extends LoggingObject {
     protected AbstractPlanInvoker(DatabaseClient client, Map<String, Object> parsedConfig) {
         this.client = client;
         String value = (String) parsedConfig.get(MarkLogicSourceConfig.KEY_COLUMN);
-        if (value != null && value.trim().length() > 0) {
+        if (StringUtils.hasText(value)) {
             this.keyColumn = value;
         } else {
             this.keyColumn = null;

src/main/java/com/marklogic/kafka/connect/source/CsvPlanInvoker.java

@@ -11,6 +11,7 @@
 import com.marklogic.kafka.connect.ConfigUtil;
 import com.marklogic.kafka.connect.MarkLogicConnectorException;
 import org.apache.kafka.connect.source.SourceRecord;
+import org.springframework.util.StringUtils;
 
 import java.io.BufferedReader;
 import java.io.IOException;
@@ -57,7 +58,7 @@ public Results invokePlan(PlanBuilder.Plan plan, String topic) {
     }
 
     private Optional<Integer> getIndexOfKeyColumn(String headerLine) {
-        if (keyColumn != null) {
+        if (StringUtils.hasText(keyColumn)) {
             ArrayNode headerNames;
             try {
                 headerNames = (ArrayNode) csvMapper.readTree(headerLine);

src/main/java/com/marklogic/kafka/connect/source/DocumentWriteOperationBuilder.java

@@ -5,6 +5,7 @@
 import com.marklogic.client.impl.DocumentWriteOperationImpl;
 import com.marklogic.client.io.DocumentMetadataHandle;
 import com.marklogic.client.io.marker.AbstractWriteHandle;
+import org.springframework.util.StringUtils;
 
 public class DocumentWriteOperationBuilder {
 
@@ -23,17 +24,17 @@ public DocumentWriteOperation build(RecordContent recordContent) {
         if (content == null) {
             throw new NullPointerException("'content' must not be null");
         }
-        if (hasText(collections)) {
+        if (StringUtils.hasText(collections)) {
             metadata.getCollections().addAll(collections.trim().split(","));
         }
-        if (hasText(permissions)) {
+        if (StringUtils.hasText(permissions)) {
             new DefaultDocumentPermissionsParser().parsePermissions(permissions.trim(), metadata.getPermissions());
         }
 
-        if (hasText(uriPrefix)) {
+        if (StringUtils.hasText(uriPrefix)) {
             uri = uriPrefix + uri;
         }
-        if (hasText(uriSuffix)) {
+        if (StringUtils.hasText(uriSuffix)) {
             uri += uriSuffix;
         }
 
@@ -54,10 +55,6 @@ protected DocumentWriteOperation build(DocumentWriteOperation.OperationType oper
         return new DocumentWriteOperationImpl(operationType, uri, metadata, content);
     }
 
-    private boolean hasText(String val) {
-        return val != null && val.trim().length() > 0;
-    }
-
     public DocumentWriteOperationBuilder withUriPrefix(String uriPrefix) {
         this.uriPrefix = uriPrefix;
         return this;

src/main/java/com/marklogic/kafka/connect/source/DslQueryHandler.java

@@ -42,7 +42,7 @@ protected String appendConstraintAndOrderByToQuery(String previousMaxConstraintC
         String constrainedDsl = userDslQuery;
         if (StringUtils.hasText(constraintColumnName)) {
             String constraintPhrase = "";
-            if (previousMaxConstraintColumnValue != null) {
+            if (StringUtils.hasText(previousMaxConstraintColumnValue)) {
                 String sanitizedValue = previousMaxConstraintColumnValue.replaceAll(VALUE_SANITIZATION_PATTERN, "");
                 constraintPhrase = String.format(".where(op.gt(op.col('%s'), '%s'))", constraintColumnName, sanitizedValue);
             }

src/main/java/com/marklogic/kafka/connect/source/JsonPlanInvoker.java

@@ -5,6 +5,7 @@
 import com.marklogic.client.expression.PlanBuilder;
 import com.marklogic.client.io.JacksonHandle;
 import org.apache.kafka.connect.source.SourceRecord;
+import org.springframework.util.StringUtils;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -31,7 +32,7 @@ public Results invokePlan(PlanBuilder.Plan plan, String topic) {
     }
 
     private String getKeyValueFromRow(JsonNode row) {
-        if (keyColumn != null && row.has(keyColumn)) {
+        if (StringUtils.hasText(keyColumn) && row.has(keyColumn)) {
             JsonNode column = row.get(keyColumn);
             return this.includeColumnTypes ? column.get("value").asText() : column.asText();
         }

src/main/java/com/marklogic/kafka/connect/source/SerializedQueryHandler.java

@@ -58,7 +58,7 @@ protected void appendConstraintAndOrderByToQuery(JsonNode currentSerializedQuery
         if (StringUtils.hasText(constraintColumnName)) {
             ObjectNode orderByNode = buildOrderByNode(true);
             ArrayNode rootArgsArray = (ArrayNode) currentSerializedQuery.get(OPTIC_PLAN_ROOT_NODE).get("args");
-            if (previousMaxConstraintColumnValue != null) {
+            if (StringUtils.hasText(previousMaxConstraintColumnValue)) {
                 ObjectNode constraintNode = buildConstraintNode(previousMaxConstraintColumnValue);
                 rootArgsArray.add(constraintNode);
             }

src/main/java/com/marklogic/kafka/connect/source/XmlPlanInvoker.java

@@ -5,6 +5,7 @@
 import com.marklogic.client.io.DOMHandle;
 import com.marklogic.kafka.connect.MarkLogicConnectorException;
 import org.apache.kafka.connect.source.SourceRecord;
+import org.springframework.util.StringUtils;
 import org.w3c.dom.Element;
 import org.w3c.dom.NamedNodeMap;
 import org.w3c.dom.Node;
@@ -65,7 +66,7 @@ private List<SourceRecord> convertRowsToSourceRecords(DOMHandle result, String t
     }
 
     private String getKeyFromRow(Node row) {
-        if (keyColumn != null) {
+        if (StringUtils.hasText(keyColumn)) {
             NodeList columns = row.getChildNodes();
             int len = columns.getLength();
             for (int j = 0; j < len; j++) {