From 8a8a3d4535e6f1acc220db1e85c19ca8f8448069 Mon Sep 17 00:00:00 2001 From: Kenneth Knowles Date: Wed, 23 Mar 2016 09:29:45 -0700 Subject: [PATCH 1/2] Replace unambiguous of `throw Throwables.propagate` with definition In the SDK the path taken by Throwables.propagate is always statically known, and the inlined logic is more explicit and readable: - If an exception e is already a checked exception, Throwables.propagate(e) is the same as `throw new RuntimeException(e)`. - If an exception e is already a RuntimeException or Error, Throwables.propagate(e) is the same as `throw e`. --- .../sdk/options/PipelineOptionsFactory.java | 4 +-- .../runners/inprocess/InProcessCreate.java | 3 +- .../dataflow/sdk/util/MutationDetectors.java | 3 +- .../sdk/testing/RestoreSystemProperties.java | 4 +-- .../cloud/dataflow/sdk/util/GcsUtilTest.java | 28 +++++++++---------- 5 files changed, 19 insertions(+), 23 deletions(-) diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/PipelineOptionsFactory.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/PipelineOptionsFactory.java index 4781d1c829df..60dbd3f6b453 100644 --- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/PipelineOptionsFactory.java +++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/PipelineOptionsFactory.java @@ -602,7 +602,7 @@ static synchronized Registration validateWellForm COMBINED_CACHE.put(combinedPipelineOptionsInterfaces, new Registration(allProxyClass, propertyDescriptors)); } catch (IntrospectionException e) { - throw Throwables.propagate(e); + throw new RuntimeException(e); } } @@ -617,7 +617,7 @@ static synchronized Registration validateWellForm INTERFACE_CACHE.put(iface, new Registration(proxyClass, propertyDescriptors)); } catch (IntrospectionException e) { - throw Throwables.propagate(e); + throw new RuntimeException(e); } } @SuppressWarnings("unchecked") diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/runners/inprocess/InProcessCreate.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/runners/inprocess/InProcessCreate.java index 9023b7b2dc4b..36e5a31861bc 100644 --- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/runners/inprocess/InProcessCreate.java +++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/runners/inprocess/InProcessCreate.java @@ -29,7 +29,6 @@ import com.google.cloud.dataflow.sdk.values.PInput; 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 com.google.common.collect.Iterators; import com.google.common.collect.PeekingIterator; @@ -75,7 +74,7 @@ public PCollection apply(PInput input) { try { source = new InMemorySource<>(original.getElements(), elementCoder); } catch (IOException e) { - throw Throwables.propagate(e); + throw new RuntimeException(e); } PCollection result = input.getPipeline().apply(Read.from(source)); result.setCoder(elementCoder); diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/MutationDetectors.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/MutationDetectors.java index 412e3eb72520..7558a0e19ad1 100644 --- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/MutationDetectors.java +++ b/sdks/java/core/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); } } diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/testing/RestoreSystemProperties.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/testing/RestoreSystemProperties.java index 03bc6a530c99..a89be3edd75e 100644 --- a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/testing/RestoreSystemProperties.java +++ b/sdks/java/core/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); } } } diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/util/GcsUtilTest.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/util/GcsUtilTest.java index e7cd7d7c22c8..29bbc4d18f25 100644 --- a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/util/GcsUtilTest.java +++ b/sdks/java/core/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; @@ -141,20 +140,21 @@ public void testMultipleThreadsCanCompleteOutOfOrderWithDefaultThreadPool() thro for (int i = 0; i < numThreads; i++) { final int currentLatch = i; countDownLatches[i] = new CountDownLatch(1); - executorService.execute(new Runnable() { - @Override - public void run() { - // Wait for latch N and then release latch N - 1 - try { - countDownLatches[currentLatch].await(); - if (currentLatch > 0) { - countDownLatches[currentLatch - 1].countDown(); + executorService.execute( + new Runnable() { + @Override + public void run() { + // Wait for latch N and then release latch N - 1 + try { + countDownLatches[currentLatch].await(); + if (currentLatch > 0) { + countDownLatches[currentLatch - 1].countDown(); + } + } catch (InterruptedException e) { + throw new RuntimeException(e); + } } - } catch (InterruptedException e) { - throw Throwables.propagate(e); - } - } - }); + }); } // Release the last latch starting the chain reaction. From 5ff607e3bb51475be1a64264cc0a3d4d87e7ff59 Mon Sep 17 00:00:00 2001 From: Kenneth Knowles Date: Thu, 31 Mar 2016 09:41:51 -0700 Subject: [PATCH 2/2] Re-interrupt thread when handling InterruptedException --- .../java/com/google/cloud/dataflow/sdk/util/GcsUtilTest.java | 1 + 1 file changed, 1 insertion(+) diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/util/GcsUtilTest.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/util/GcsUtilTest.java index 29bbc4d18f25..65fb83812606 100644 --- a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/util/GcsUtilTest.java +++ b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/util/GcsUtilTest.java @@ -151,6 +151,7 @@ public void run() { countDownLatches[currentLatch - 1].countDown(); } } catch (InterruptedException e) { + Thread.currentThread().interrupt(); throw new RuntimeException(e); } }