DataflowJavaSDK: remove all uses of Throwables.propagate

examples/src/main/java/com/google/cloud/dataflow/examples/common/DataflowExampleUtils.java

@@ -47,7 +47,6 @@
 import com.google.cloud.dataflow.sdk.values.PBegin;
 import com.google.cloud.dataflow.sdk.values.PCollection;
 import com.google.common.base.Strings;
-import com.google.common.base.Throwables;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Sets;
 
@@ -110,7 +109,7 @@ public void setup() throws IOException {
     } catch (InterruptedException e) {
       // Ignore InterruptedException
     }
-    Throwables.propagate(lastException);
+    throw new RuntimeException(lastException);
   }
 
   /**

sdk/src/main/java/com/google/cloud/dataflow/sdk/io/PubsubIO.java

@@ -50,7 +50,6 @@
 import com.google.cloud.dataflow.sdk.values.PDone;
 import com.google.cloud.dataflow.sdk.values.PInput;
 import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Throwables;
 import com.google.common.collect.ImmutableMap;
 
 import org.joda.time.Duration;
@@ -814,7 +813,7 @@ public void processElement(ProcessContext c) throws IOException {
             }
           }
           if (finallyBlockException != null) {
-            Throwables.propagate(finallyBlockException);
+            throw new RuntimeException(finallyBlockException);
           }
 
           for (PubsubMessage message : messages) {

sdk/src/main/java/com/google/cloud/dataflow/sdk/options/PipelineOptionsFactory.java

@@ -31,7 +31,6 @@
 import com.google.common.base.Preconditions;
 import com.google.common.base.Predicate;
 import com.google.common.base.Strings;
-import com.google.common.base.Throwables;
 import com.google.common.collect.Collections2;
 import com.google.common.collect.FluentIterable;
 import com.google.common.collect.ImmutableListMultimap;
@@ -624,7 +623,7 @@ static synchronized <T extends PipelineOptions> Registration<T> validateWellForm
         COMBINED_CACHE.put(combinedPipelineOptionsInterfaces,
             new Registration<T>(allProxyClass, propertyDescriptors));
       } catch (IntrospectionException e) {
-        throw Throwables.propagate(e);
+        throw new RuntimeException(e);
       }
     }
 
@@ -639,7 +638,7 @@ static synchronized <T extends PipelineOptions> Registration<T> validateWellForm
         INTERFACE_CACHE.put(iface,
             new Registration<T>(proxyClass, propertyDescriptors));
       } catch (IntrospectionException e) {
-        throw Throwables.propagate(e);
+        throw new RuntimeException(e);
       }
     }
     @SuppressWarnings("unchecked")
@@ -1038,7 +1037,7 @@ private static List<PropertyDescriptor> validateClass(Class<? extends PipelineOp
       methods.add(klass.getMethod("cloneAs", Class.class));
       methods.add(klass.getMethod("populateDisplayData", DisplayData.Builder.class));
     } catch (NoSuchMethodException | SecurityException e) {
-      throw Throwables.propagate(e);
+      throw new RuntimeException(e);
     }
 
     // Verify that there are no methods with the same name with two different return types.

sdk/src/main/java/com/google/cloud/dataflow/sdk/runners/DataflowPipelineJob.java

@@ -36,7 +36,6 @@
 import com.google.cloud.dataflow.sdk.util.MapAggregatorValues;
 import com.google.cloud.dataflow.sdk.util.MonitoringUtil;
 import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Throwables;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -348,7 +347,7 @@ private boolean nextBackOff(Sleeper sleeper, BackOff backoff) {
     try {
       return BackOffUtils.next(sleeper, backoff);
     } catch (InterruptedException | IOException e) {
-      throw Throwables.propagate(e);
+      throw new RuntimeException(e);
     }
   }
 

sdk/src/main/java/com/google/cloud/dataflow/sdk/runners/inprocess/InProcessCreate.java

@@ -32,7 +32,6 @@
 import com.google.cloud.dataflow.sdk.values.POutput;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Optional;
-import com.google.common.base.Throwables;
 import com.google.common.collect.ImmutableList;
 
 import java.io.IOException;
@@ -91,7 +90,7 @@ public PCollection<T> apply(PInput input) {
     try {
       source = InMemorySource.fromIterable(original.getElements(), elementCoder);
     } catch (IOException e) {
-      throw Throwables.propagate(e);
+      throw new RuntimeException(e);
     }
     PCollection<T> result = input.getPipeline().apply(Read.from(source));
     result.setCoder(elementCoder);

sdk/src/main/java/com/google/cloud/dataflow/sdk/runners/inprocess/InProcessPipelineRunner.java

@@ -47,7 +47,6 @@
 import com.google.cloud.dataflow.sdk.values.PInput;
 import com.google.cloud.dataflow.sdk.values.POutput;
 import com.google.cloud.dataflow.sdk.values.PValue;
-import com.google.common.base.Throwables;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
@@ -239,7 +238,7 @@ public InProcessPipelineResult run(Pipeline pipeline) {
       } catch (UserCodeException userException) {
         throw new PipelineExecutionException(userException.getCause());
       } catch (Throwable t) {
-        Throwables.propagate(t);
+        throw new RuntimeException(t);
       }
     }
     return result;

sdk/src/main/java/com/google/cloud/dataflow/sdk/runners/inprocess/InProcessSideInputContainer.java

@@ -26,7 +26,6 @@
 import com.google.cloud.dataflow.sdk.util.WindowingStrategy;
 import com.google.cloud.dataflow.sdk.values.PCollectionView;
 import com.google.common.base.MoreObjects;
-import com.google.common.base.Throwables;
 import com.google.common.cache.CacheBuilder;
 import com.google.common.cache.CacheLoader;
 import com.google.common.cache.LoadingCache;
@@ -169,7 +168,7 @@ private void updatePCollectionViewWindowValues(
         future.set(Collections.<WindowedValue<?>>emptyList());
       }
     } catch (ExecutionException e) {
-      Throwables.propagate(e.getCause());
+      throw new RuntimeException(e.getCause());
     }
   }
 

sdk/src/main/java/com/google/cloud/dataflow/sdk/runners/inprocess/TransformExecutor.java

@@ -20,7 +20,6 @@
 import com.google.cloud.dataflow.sdk.runners.inprocess.InProcessPipelineRunner.CommittedBundle;
 import com.google.cloud.dataflow.sdk.transforms.AppliedPTransform;
 import com.google.cloud.dataflow.sdk.util.WindowedValue;
-import com.google.common.base.Throwables;
 
 import java.util.ArrayList;
 import java.util.Collection;
@@ -116,7 +115,10 @@ public InProcessTransformResult call() {
       return result;
     } catch (Throwable t) {
       onComplete.handleThrowable(inputBundle, t);
-      throw Throwables.propagate(t);
+      if (t instanceof RuntimeException) {
+        throw (RuntimeException) t;
+      }
+      throw new RuntimeException(t);
     } finally {
       transformEvaluationState.complete(this);
     }

sdk/src/main/java/com/google/cloud/dataflow/sdk/testing/TestDataflowPipelineRunner.java

@@ -125,7 +125,7 @@ public void process(List<JobMessage> messages) {
                   try {
                     job.cancel();
                   } catch (Exception e) {
-                    throw Throwables.propagate(e);
+                    throw new RuntimeException(e);
                   }
                 }
               }
@@ -149,7 +149,7 @@ public void process(List<JobMessage> messages) {
       }
     } catch (Exception e) {
       Throwables.propagateIfPossible(e);
-      throw Throwables.propagate(e);
+      throw new RuntimeException(e);
     }
     return job;
   }

sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/DoFnReflector.java

@@ -32,7 +32,6 @@
 import com.google.cloud.dataflow.sdk.values.TypeDescriptor;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Function;
-import com.google.common.base.Throwables;
 import com.google.common.collect.FluentIterable;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.reflect.TypeParameter;
@@ -477,7 +476,7 @@ private <InputT, OutputT> void invoke(Method m,
         throw UserCodeException.wrap(e.getCause());
       } catch (IllegalAccessException | IllegalArgumentException e) {
         // Exception in our code.
-        throw Throwables.propagate(e);
+        throw new RuntimeException(e);
       }
     }
   }

sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/IntraBundleParallelization.java

@@ -217,7 +217,7 @@ public void processElement(final ProcessContext c) throws Exception {
       }
 
       if (failure.get() != null) {
-        throw Throwables.propagate(failure.get());
+        throw new RuntimeException(failure.get());
       }
 
       executor.submit(new Runnable() {
@@ -242,7 +242,7 @@ public void finishBundle(Context c) throws Exception {
       // processElement calls have finished.
       workTickets.acquire(maxParallelism);
       if (failure.get() != null) {
-        throw Throwables.propagate(failure.get());
+        throw new RuntimeException(failure.get());
       }
       doFn.finishBundle(c);
     }

sdk/src/main/java/com/google/cloud/dataflow/sdk/util/BigQueryTableInserter.java

@@ -35,7 +35,6 @@
 import com.google.cloud.hadoop.util.ApiErrorExtractor;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
-import com.google.common.base.Throwables;
 import com.google.common.util.concurrent.MoreExecutors;
 
 import org.slf4j.Logger;
@@ -272,7 +271,7 @@ public List<TableDataInsertAllResponse.InsertErrors> call() throws IOException {
       } catch (InterruptedException e) {
         throw new IOException("Interrupted while inserting " + rowsToPublish);
       } catch (ExecutionException e) {
-        Throwables.propagate(e.getCause());
+        throw new RuntimeException(e.getCause());
       }
 
       if (!allErrors.isEmpty() && !backoff.atMaxAttempts()) {

sdk/src/main/java/com/google/cloud/dataflow/sdk/util/MutationDetectors.java

@@ -18,7 +18,6 @@
 
 import com.google.cloud.dataflow.sdk.coders.Coder;
 import com.google.cloud.dataflow.sdk.coders.CoderException;
-import com.google.common.base.Throwables;
 
 import java.util.Arrays;
 import java.util.Objects;
@@ -113,7 +112,7 @@ public void verifyUnmodified() {
       try {
         verifyUnmodifiedThrowingCheckedExceptions();
       } catch (CoderException exn) {
-        Throwables.propagate(exn);
+        throw new RuntimeException(exn);
       }
     }
 

sdk/src/test/java/com/google/cloud/dataflow/sdk/testing/RestoreSystemProperties.java

@@ -16,8 +16,6 @@
 
 package com.google.cloud.dataflow.sdk.testing;
 
-import com.google.common.base.Throwables;
-
 import org.junit.rules.ExternalResource;
 import org.junit.rules.TestRule;
 
@@ -45,7 +43,7 @@ protected void after() {
       System.getProperties().clear();
       System.getProperties().load(bais);
     } catch (IOException e) {
-      throw Throwables.propagate(e);
+      throw new RuntimeException(e);
     }
   }
 }

sdk/src/test/java/com/google/cloud/dataflow/sdk/util/GcsUtilTest.java

@@ -52,7 +52,6 @@
 import com.google.cloud.dataflow.sdk.util.gcsfs.GcsPath;
 import com.google.cloud.hadoop.gcsio.GoogleCloudStorageReadChannel;
 import com.google.cloud.hadoop.util.ClientRequestHelper;
-import com.google.common.base.Throwables;
 import com.google.common.collect.ImmutableList;
 
 import org.junit.Rule;
@@ -151,7 +150,7 @@ public void run() {
               countDownLatches[currentLatch - 1].countDown();
             }
           } catch (InterruptedException e) {
-            throw Throwables.propagate(e);
+            throw new RuntimeException(e);
           }
         }
       });

sdk/src/test/java/com/google/cloud/dataflow/sdk/util/ReduceFnTester.java

@@ -56,7 +56,6 @@
 import com.google.common.base.Function;
 import com.google.common.base.MoreObjects;
 import com.google.common.base.Preconditions;
-import com.google.common.base.Throwables;
 import com.google.common.collect.FluentIterable;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
@@ -406,7 +405,7 @@ public WindowedValue<InputT> apply(TimestampedValue<InputT> input) {
                   windowFn, value, timestamp, Arrays.asList(GlobalWindow.INSTANCE)));
               return WindowedValue.of(value, timestamp, windows, PaneInfo.NO_FIRING);
             } catch (Exception e) {
-              throw Throwables.propagate(e);
+              throw new RuntimeException(e);
             }
           }
         }));

sdk/src/test/java/com/google/cloud/dataflow/sdk/util/TriggerTester.java

@@ -42,7 +42,6 @@
 import com.google.cloud.dataflow.sdk.values.TimestampedValue;
 import com.google.common.base.MoreObjects;
 import com.google.common.base.Preconditions;
-import com.google.common.base.Throwables;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 
@@ -254,7 +253,7 @@ public final void injectElements(Collection<TimestampedValue<InputT>> values) th
 
         windowedValues.add(WindowedValue.of(value, timestamp, assignedWindows, PaneInfo.NO_FIRING));
       } catch (Exception e) {
-        throw Throwables.propagate(e);
+        throw new RuntimeException(e);
       }
     }