Merge pull request #146 from marklogic-community/feature/more-null-handling

rjrudin · web-flow · commit f5a927f3007e · 2023-01-27T10:49:55.000-08:00
More StringUtils sanity checking
diff --git a/src/main/java/com/marklogic/kafka/connect/DefaultDatabaseClientConfigBuilder.java b/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()));
         }
 
diff --git a/src/main/java/com/marklogic/kafka/connect/sink/WriteBatcherSinkTask.java b/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 " +
diff --git a/src/main/java/com/marklogic/kafka/connect/source/AbstractPlanInvoker.java b/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;
diff --git a/src/main/java/com/marklogic/kafka/connect/source/CsvPlanInvoker.java b/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);
diff --git a/src/main/java/com/marklogic/kafka/connect/source/DocumentWriteOperationBuilder.java b/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;
diff --git a/src/main/java/com/marklogic/kafka/connect/source/DslQueryHandler.java b/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);
             }
diff --git a/src/main/java/com/marklogic/kafka/connect/source/JsonPlanInvoker.java b/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();
         }
diff --git a/src/main/java/com/marklogic/kafka/connect/source/SerializedQueryHandler.java b/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);
             }
diff --git a/src/main/java/com/marklogic/kafka/connect/source/XmlPlanInvoker.java b/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++) {

Original file line number	Diff line number	Diff line change
`@@ -5,6 +5,7 @@`
`5`	`5`	`import com.marklogic.client.impl.DocumentWriteOperationImpl;`
`6`	`6`	`import com.marklogic.client.io.DocumentMetadataHandle;`
`7`	`7`	`import com.marklogic.client.io.marker.AbstractWriteHandle;`
	`8`	`+import org.springframework.util.StringUtils;`
`8`	`9`
`9`	`10`	`public class DocumentWriteOperationBuilder {`
`10`	`11`
`@@ -23,17 +24,17 @@ public DocumentWriteOperation build(RecordContent recordContent) {`
`23`	`24`	`if (content == null) {`
`24`	`25`	`throw new NullPointerException("'content' must not be null");`
`25`	`26`	`}`
`26`		`- if (hasText(collections)) {`
	`27`	`+ if (StringUtils.hasText(collections)) {`
`27`	`28`	`metadata.getCollections().addAll(collections.trim().split(","));`
`28`	`29`	`}`
`29`		`- if (hasText(permissions)) {`
	`30`	`+ if (StringUtils.hasText(permissions)) {`
`30`	`31`	`new DefaultDocumentPermissionsParser().parsePermissions(permissions.trim(), metadata.getPermissions());`
`31`	`32`	`}`
`32`	`33`
`33`		`- if (hasText(uriPrefix)) {`
	`34`	`+ if (StringUtils.hasText(uriPrefix)) {`
`34`	`35`	`uri = uriPrefix + uri;`
`35`	`36`	`}`
`36`		`- if (hasText(uriSuffix)) {`
	`37`	`+ if (StringUtils.hasText(uriSuffix)) {`
`37`	`38`	`uri += uriSuffix;`
`38`	`39`	`}`
`39`	`40`
`@@ -54,10 +55,6 @@ protected DocumentWriteOperation build(DocumentWriteOperation.OperationType oper`
`54`	`55`	`return new DocumentWriteOperationImpl(operationType, uri, metadata, content);`
`55`	`56`	`}`
`56`	`57`
`57`		`- private boolean hasText(String val) {`
`58`		`- return val != null && val.trim().length() > 0;`
`59`		`- }`
`60`		`-`
`61`	`58`	`public DocumentWriteOperationBuilder withUriPrefix(String uriPrefix) {`
`62`	`59`	`this.uriPrefix = uriPrefix;`
`63`	`60`	`return this;`
Original file line number	Diff line number	Diff line change
`@@ -42,7 +42,7 @@ protected String appendConstraintAndOrderByToQuery(String previousMaxConstraintC`
`42`	`42`	`String constrainedDsl = userDslQuery;`
`43`	`43`	`if (StringUtils.hasText(constraintColumnName)) {`
`44`	`44`	`String constraintPhrase = "";`
`45`		`- if (previousMaxConstraintColumnValue != null) {`
	`45`	`+ if (StringUtils.hasText(previousMaxConstraintColumnValue)) {`
`46`	`46`	`String sanitizedValue = previousMaxConstraintColumnValue.replaceAll(VALUE_SANITIZATION_PATTERN, "");`
`47`	`47`	`constraintPhrase = String.format(".where(op.gt(op.col('%s'), '%s'))", constraintColumnName, sanitizedValue);`
`48`	`48`	`}`
Original file line number	Diff line number	Diff line change
`@@ -5,6 +5,7 @@`
`5`	`5`	`import com.marklogic.client.expression.PlanBuilder;`
`6`	`6`	`import com.marklogic.client.io.JacksonHandle;`
`7`	`7`	`import org.apache.kafka.connect.source.SourceRecord;`
	`8`	`+import org.springframework.util.StringUtils;`
`8`	`9`
`9`	`10`	`import java.util.ArrayList;`
`10`	`11`	`import java.util.List;`
`@@ -31,7 +32,7 @@ public Results invokePlan(PlanBuilder.Plan plan, String topic) {`
`31`	`32`	`}`
`32`	`33`
`33`	`34`	`private String getKeyValueFromRow(JsonNode row) {`
`34`		`- if (keyColumn != null && row.has(keyColumn)) {`
	`35`	`+ if (StringUtils.hasText(keyColumn) && row.has(keyColumn)) {`
`35`	`36`	`JsonNode column = row.get(keyColumn);`
`36`	`37`	`return this.includeColumnTypes ? column.get("value").asText() : column.asText();`
`37`	`38`	`}`
Original file line number	Diff line number	Diff line change
`@@ -58,7 +58,7 @@ protected void appendConstraintAndOrderByToQuery(JsonNode currentSerializedQuery`
`58`	`58`	`if (StringUtils.hasText(constraintColumnName)) {`
`59`	`59`	`ObjectNode orderByNode = buildOrderByNode(true);`
`60`	`60`	`ArrayNode rootArgsArray = (ArrayNode) currentSerializedQuery.get(OPTIC_PLAN_ROOT_NODE).get("args");`
`61`		`- if (previousMaxConstraintColumnValue != null) {`
	`61`	`+ if (StringUtils.hasText(previousMaxConstraintColumnValue)) {`
`62`	`62`	`ObjectNode constraintNode = buildConstraintNode(previousMaxConstraintColumnValue);`
`63`	`63`	`rootArgsArray.add(constraintNode);`
`64`	`64`	`}`