Fix ingestion-sink cross-project inserts

relud · relud · commit 37e6273a283a · 2019-08-14T19:28:38.000-07:00
Update google cloud library to 1.61.0 to match beam 2.13.0 and pick up googleapis/google-cloud-java#4196 Emit stack traces in ingestion-sink
diff --git a/ingestion-sink/src/main/java/com/mozilla/telemetry/ingestion/io/BigQuery.java b/ingestion-sink/src/main/java/com/mozilla/telemetry/ingestion/io/BigQuery.java
@@ -137,7 +137,7 @@ private synchronized Optional<CompletableFuture<TableRow>> add(TableRow row) {
         int index = newSize - 1;
         return Optional.of(result.thenApplyAsync(r -> {
           List<BigQueryError> errors = r.getErrorsFor(index);
-          if (!errors.isEmpty()) {
+          if (errors != null && !errors.isEmpty()) {
             throw new WriteErrors(errors);
           }
           return row;
diff --git a/ingestion-sink/src/main/java/com/mozilla/telemetry/ingestion/io/Pubsub.java b/ingestion-sink/src/main/java/com/mozilla/telemetry/ingestion/io/Pubsub.java
@@ -10,6 +10,7 @@
 import com.google.pubsub.v1.PubsubMessage;
 import java.util.concurrent.CompletableFuture;
 import java.util.function.Function;
+import java.util.logging.Logger;
 
 public class Pubsub {
 
@@ -18,6 +19,8 @@ private Pubsub() {
 
   public static class Read {
 
+    static Logger logger = Logger.getLogger("Pubsub.Read");
+
     @VisibleForTesting
     Subscriber subscriber;
 
@@ -30,6 +33,7 @@ public Read(String subscriptionName, Function<PubsubMessage, CompletableFuture<?
                 if (exception == null) {
                   consumer.ack();
                 } else {
+                  logger.warning("Exception while attempting to deliver message:", exception);
                   consumer.nack();
                 }
               })))
diff --git a/pom.xml b/pom.xml
@@ -27,7 +27,7 @@
 
         <!-- Keep these dependency versions in sync with those pulled in by beam;
              check https://mvnrepository.com/artifact/org.apache.beam -->
-        <google-cloud.version>1.49.0</google-cloud.version>
+        <google-cloud.version>1.61.0</google-cloud.version>
         <jackson.version>2.9.9</jackson.version>
     </properties>
 

Original file line number	Diff line number	Diff line change
`@@ -137,7 +137,7 @@ private synchronized Optional<CompletableFuture<TableRow>> add(TableRow row) {`
`137`	`137`	`int index = newSize - 1;`
`138`	`138`	`return Optional.of(result.thenApplyAsync(r -> {`
`139`	`139`	`List<BigQueryError> errors = r.getErrorsFor(index);`
`140`		`- if (!errors.isEmpty()) {`
	`140`	`+ if (errors != null && !errors.isEmpty()) {`
`141`	`141`	`throw new WriteErrors(errors);`
`142`	`142`	`}`
`143`	`143`	`return row;`