Fix for incorrect rpc url

apache · Mar 18, 2021 · a923846 · a923846
1 parent aeb471e
commit a923846
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 1 deletion.
diff --git a/...va/src/main/java/org/apache/beam/examples/complete/datatokenization/DataTokenization.java b/...va/src/main/java/org/apache/beam/examples/complete/datatokenization/DataTokenization.java
@@ -55,6 +55,7 @@
 import org.apache.beam.sdk.values.Row;
 import org.apache.beam.sdk.values.TupleTag;
 import org.apache.beam.sdk.values.TypeDescriptors;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Strings;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -275,7 +276,7 @@ public static PipelineResult run(DataTokenizationOptions options) {
                   .via(
                       (Row errRow) ->
                           FailsafeElement.of(errRow.getString("line"), errRow.getString("line"))
-                              .setErrorMessage(errRow.getString("err"))))
+                              .setErrorMessage(Strings.nullToEmpty(errRow.getString("err")))))
           .apply(
               "WriteCsvConversionErrorsToFS",
               ErrorConverters.WriteStringMessageErrorsAsCsv.newBuilder()

diff --git a/...in/java/org/apache/beam/examples/complete/datatokenization/transforms/DataProtectors.java b/...in/java/org/apache/beam/examples/complete/datatokenization/transforms/DataProtectors.java
@@ -63,6 +63,7 @@
 import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Throwables;
 import org.apache.commons.io.IOUtils;
 import org.apache.http.HttpEntity;
+import org.apache.http.HttpStatus;
 import org.apache.http.client.methods.CloseableHttpResponse;
 import org.apache.http.client.methods.HttpPost;
 import org.apache.http.entity.ByteArrayEntity;
@@ -291,6 +292,10 @@ private ArrayList<Row> getTokenizedRow(Iterable<Row> inputRows) throws IOExcepti
       CloseableHttpResponse response =
           sendRpc(formatJsonsToRpcBatch(rowsToJsons(inputRows)).getBytes(Charset.defaultCharset()));
 
+      if (response.getStatusLine().getStatusCode() != HttpStatus.SC_OK) {
+        LOG.error("Send to RPC '{}' failed with '{}'", this.rpcURI, response.getStatusLine());
+      }
+
       String tokenizedData =
           IOUtils.toString(response.getEntity().getContent(), StandardCharsets.UTF_8);