You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by dh...@apache.org on 2016/05/03 00:33:10 UTC
[1/2] incubator-beam git commit: Closes #270
Repository: incubator-beam
Updated Branches:
refs/heads/master 1262819ac -> 3ebdf4886
Closes #270
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/3ebdf488
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/3ebdf488
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/3ebdf488
Branch: refs/heads/master
Commit: 3ebdf48860017897d5a41340bc2670782ff4c68f
Parents: 1262819 454031a
Author: Dan Halperin <dh...@google.com>
Authored: Mon May 2 15:32:16 2016 -0700
Committer: Dan Halperin <dh...@google.com>
Committed: Mon May 2 15:32:16 2016 -0700
----------------------------------------------------------------------
.../beam/examples/common/DataflowExampleUtils.java | 3 +--
runners/direct-java/pom.xml | 14 --------------
.../direct/ImmutabilityCheckingBundleFactory.java | 3 +--
.../beam/runners/direct/InProcessPipelineRunner.java | 6 ++++--
.../runners/direct/InProcessSideInputContainer.java | 3 +--
.../apache/beam/runners/direct/TransformExecutor.java | 7 ++++---
.../beam/runners/dataflow/DataflowPipelineJob.java | 3 +--
.../src/main/resources/beam/checkstyle.xml | 6 ++++++
.../main/java/org/apache/beam/sdk/io/PubsubIO.java | 3 +--
.../beam/sdk/options/PipelineOptionsFactory.java | 4 ++--
.../org/apache/beam/sdk/transforms/DoFnReflector.java | 3 +--
.../sdk/transforms/IntraBundleParallelization.java | 4 ++--
.../apache/beam/sdk/util/BigQueryTableInserter.java | 3 +--
.../java/org/apache/beam/sdk/util/ReduceFnTester.java | 3 +--
.../java/org/apache/beam/sdk/util/TriggerTester.java | 4 +---
.../java/org/apache/beam/sdk/io/kafka/KafkaIO.java | 3 +--
16 files changed, 28 insertions(+), 44 deletions(-)
----------------------------------------------------------------------
[2/2] incubator-beam git commit: beam-wide: blacklist
Throwables.propagate and remove uses
Posted by dh...@apache.org.
beam-wide: blacklist Throwables.propagate and remove uses
This is a forward-port of https://github.com/GoogleCloudPlatform/DataflowJavaSDK/pull/232
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/454031af
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/454031af
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/454031af
Branch: refs/heads/master
Commit: 454031af48b045cdbe3f689c216be62fc6d6a86d
Parents: 1262819
Author: Dan Halperin <dh...@google.com>
Authored: Mon May 2 13:25:40 2016 -0700
Committer: Dan Halperin <dh...@google.com>
Committed: Mon May 2 15:32:16 2016 -0700
----------------------------------------------------------------------
.../beam/examples/common/DataflowExampleUtils.java | 3 +--
runners/direct-java/pom.xml | 14 --------------
.../direct/ImmutabilityCheckingBundleFactory.java | 3 +--
.../beam/runners/direct/InProcessPipelineRunner.java | 6 ++++--
.../runners/direct/InProcessSideInputContainer.java | 3 +--
.../apache/beam/runners/direct/TransformExecutor.java | 7 ++++---
.../beam/runners/dataflow/DataflowPipelineJob.java | 3 +--
.../src/main/resources/beam/checkstyle.xml | 6 ++++++
.../main/java/org/apache/beam/sdk/io/PubsubIO.java | 3 +--
.../beam/sdk/options/PipelineOptionsFactory.java | 4 ++--
.../org/apache/beam/sdk/transforms/DoFnReflector.java | 3 +--
.../sdk/transforms/IntraBundleParallelization.java | 4 ++--
.../apache/beam/sdk/util/BigQueryTableInserter.java | 3 +--
.../java/org/apache/beam/sdk/util/ReduceFnTester.java | 3 +--
.../java/org/apache/beam/sdk/util/TriggerTester.java | 4 +---
.../java/org/apache/beam/sdk/io/kafka/KafkaIO.java | 3 +--
16 files changed, 28 insertions(+), 44 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/454031af/examples/java/src/main/java/org/apache/beam/examples/common/DataflowExampleUtils.java
----------------------------------------------------------------------
diff --git a/examples/java/src/main/java/org/apache/beam/examples/common/DataflowExampleUtils.java b/examples/java/src/main/java/org/apache/beam/examples/common/DataflowExampleUtils.java
index 7ac71d3..fb4f3bf 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/common/DataflowExampleUtils.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/common/DataflowExampleUtils.java
@@ -51,7 +51,6 @@ import com.google.api.services.pubsub.Pubsub;
import com.google.api.services.pubsub.model.Subscription;
import com.google.api.services.pubsub.model.Topic;
import com.google.common.base.Strings;
-import com.google.common.base.Throwables;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import com.google.common.util.concurrent.Uninterruptibles;
@@ -116,7 +115,7 @@ public class DataflowExampleUtils {
Thread.currentThread().interrupt();
// Ignore InterruptedException
}
- Throwables.propagate(lastException);
+ throw new RuntimeException(lastException);
}
/**
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/454031af/runners/direct-java/pom.xml
----------------------------------------------------------------------
diff --git a/runners/direct-java/pom.xml b/runners/direct-java/pom.xml
index 12ba329..15a4e9c 100644
--- a/runners/direct-java/pom.xml
+++ b/runners/direct-java/pom.xml
@@ -280,20 +280,6 @@
<dependency>
<groupId>com.google.http-client</groupId>
- <artifactId>google-http-client</artifactId>
- <version>${google-clients.version}</version>
- <exclusions>
- <!-- Exclude an old version of guava that is being pulled
- in by a transitive dependency of google-api-client -->
- <exclusion>
- <groupId>com.google.guava</groupId>
- <artifactId>guava-jdk5</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
-
- <dependency>
- <groupId>com.google.http-client</groupId>
<artifactId>google-http-client-protobuf</artifactId>
<version>${google-clients.version}</version>
<exclusions>
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/454031af/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ImmutabilityCheckingBundleFactory.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ImmutabilityCheckingBundleFactory.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ImmutabilityCheckingBundleFactory.java
index 2103ad3..3b38211 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ImmutabilityCheckingBundleFactory.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ImmutabilityCheckingBundleFactory.java
@@ -31,7 +31,6 @@ import org.apache.beam.sdk.util.UserCodeException;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.values.PCollection;
-import com.google.api.client.util.Throwables;
import com.google.common.collect.HashMultimap;
import com.google.common.collect.SetMultimap;
@@ -100,7 +99,7 @@ class ImmutabilityCheckingBundleFactory implements BundleFactory {
mutationDetectors.put(
element, MutationDetectors.forValueWithCoder(element.getValue(), coder));
} catch (CoderException e) {
- throw Throwables.propagate(e);
+ throw new RuntimeException(e);
}
underlying.add(element);
return this;
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/454031af/runners/direct-java/src/main/java/org/apache/beam/runners/direct/InProcessPipelineRunner.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/InProcessPipelineRunner.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/InProcessPipelineRunner.java
index bb8c0de..19e9f47 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/InProcessPipelineRunner.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/InProcessPipelineRunner.java
@@ -50,7 +50,6 @@ import org.apache.beam.sdk.values.PInput;
import org.apache.beam.sdk.values.POutput;
import org.apache.beam.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;
@@ -256,7 +255,10 @@ public class InProcessPipelineRunner
} catch (UserCodeException userException) {
throw new PipelineExecutionException(userException.getCause());
} catch (Throwable t) {
- Throwables.propagate(t);
+ if (t instanceof RuntimeException) {
+ throw (RuntimeException) t;
+ }
+ throw new RuntimeException(t);
}
}
return result;
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/454031af/runners/direct-java/src/main/java/org/apache/beam/runners/direct/InProcessSideInputContainer.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/InProcessSideInputContainer.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/InProcessSideInputContainer.java
index f4980ef..b01cd48 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/InProcessSideInputContainer.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/InProcessSideInputContainer.java
@@ -29,7 +29,6 @@ import org.apache.beam.sdk.util.WindowingStrategy;
import org.apache.beam.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;
@@ -172,7 +171,7 @@ class InProcessSideInputContainer {
future.set(Collections.<WindowedValue<?>>emptyList());
}
} catch (ExecutionException e) {
- Throwables.propagate(e.getCause());
+ throw new RuntimeException(e.getCause());
}
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/454031af/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutor.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutor.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutor.java
index 8346e89..9e15c2a 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutor.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutor.java
@@ -23,8 +23,6 @@ import org.apache.beam.runners.direct.InProcessPipelineRunner.CommittedBundle;
import org.apache.beam.sdk.transforms.AppliedPTransform;
import org.apache.beam.sdk.util.WindowedValue;
-import com.google.common.base.Throwables;
-
import java.util.ArrayList;
import java.util.Collection;
import java.util.concurrent.Callable;
@@ -119,7 +117,10 @@ class TransformExecutor<T> implements Callable<InProcessTransformResult> {
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);
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/454031af/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineJob.java
----------------------------------------------------------------------
diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineJob.java b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineJob.java
index 69565ac..0f42148 100644
--- a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineJob.java
+++ b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineJob.java
@@ -41,7 +41,6 @@ import com.google.api.services.dataflow.model.JobMessage;
import com.google.api.services.dataflow.model.JobMetrics;
import com.google.api.services.dataflow.model.MetricUpdate;
import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Throwables;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -356,7 +355,7 @@ public class DataflowPipelineJob implements PipelineResult {
if (e instanceof InterruptedException) {
Thread.currentThread().interrupt();
}
- throw Throwables.propagate(e);
+ throw new RuntimeException(e);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/454031af/sdks/java/build-tools/src/main/resources/beam/checkstyle.xml
----------------------------------------------------------------------
diff --git a/sdks/java/build-tools/src/main/resources/beam/checkstyle.xml b/sdks/java/build-tools/src/main/resources/beam/checkstyle.xml
index 31717ff..2a4f832 100644
--- a/sdks/java/build-tools/src/main/resources/beam/checkstyle.xml
+++ b/sdks/java/build-tools/src/main/resources/beam/checkstyle.xml
@@ -59,6 +59,12 @@ page at http://checkstyle.sourceforge.net/config.html -->
<property name="severity" value="error"/>
</module>
+ <module name="RegexpSingleline">
+ <property name="format" value="Throwables.propagate\("/>
+ <property name="message" value="Throwables.propagate is deprecated"/>
+ <property name="severity" value="error"/>
+ </module>
+
<!-- Allow use of comment to suppress javadocstyle -->
<module name="SuppressionCommentFilter">
<property name="offCommentFormat" value="CHECKSTYLE.OFF\: ([\w\|]+)"/>
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/454031af/sdks/java/core/src/main/java/org/apache/beam/sdk/io/PubsubIO.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/PubsubIO.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/PubsubIO.java
index 66d1d43..4646461 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/PubsubIO.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/PubsubIO.java
@@ -51,7 +51,6 @@ import com.google.api.services.pubsub.model.PullResponse;
import com.google.api.services.pubsub.model.ReceivedMessage;
import com.google.api.services.pubsub.model.Subscription;
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 class PubsubIO {
}
}
if (finallyBlockException != null) {
- Throwables.propagate(finallyBlockException);
+ throw new RuntimeException(finallyBlockException);
}
for (PubsubMessage message : messages) {
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/454031af/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptionsFactory.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptionsFactory.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptionsFactory.java
index 5fc7312..5f2dd11 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptionsFactory.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptionsFactory.java
@@ -25,6 +25,7 @@ import org.apache.beam.sdk.runners.PipelineRunnerRegistrar;
import org.apache.beam.sdk.transforms.display.DisplayData;
import org.apache.beam.sdk.util.StringUtils;
import org.apache.beam.sdk.util.common.ReflectHelpers;
+
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Function;
import com.google.common.base.Joiner;
@@ -32,7 +33,6 @@ import com.google.common.base.Optional;
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;
@@ -991,7 +991,7 @@ public class PipelineOptionsFactory {
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.
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/454031af/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnReflector.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnReflector.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnReflector.java
index cb9c43b..452ee8e 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnReflector.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnReflector.java
@@ -34,7 +34,6 @@ import org.apache.beam.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;
@@ -479,7 +478,7 @@ public abstract class DoFnReflector {
throw UserCodeException.wrap(e.getCause());
} catch (IllegalAccessException | IllegalArgumentException e) {
// Exception in our code.
- throw Throwables.propagate(e);
+ throw new RuntimeException(e);
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/454031af/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/IntraBundleParallelization.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/IntraBundleParallelization.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/IntraBundleParallelization.java
index 4b3afb4..62c09c2 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/IntraBundleParallelization.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/IntraBundleParallelization.java
@@ -221,7 +221,7 @@ public class IntraBundleParallelization {
}
if (failure.get() != null) {
- throw Throwables.propagate(failure.get());
+ throw new RuntimeException(failure.get());
}
executor.submit(new Runnable() {
@@ -246,7 +246,7 @@ public class IntraBundleParallelization {
// processElement calls have finished.
workTickets.acquire(maxParallelism);
if (failure.get() != null) {
- throw Throwables.propagate(failure.get());
+ throw new RuntimeException(failure.get());
}
doFn.finishBundle(c);
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/454031af/sdks/java/core/src/main/java/org/apache/beam/sdk/util/BigQueryTableInserter.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/BigQueryTableInserter.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/BigQueryTableInserter.java
index 86a0b5b..0493f1c 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/BigQueryTableInserter.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/BigQueryTableInserter.java
@@ -37,7 +37,6 @@ import com.google.api.services.bigquery.model.TableSchema;
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;
@@ -275,7 +274,7 @@ public class BigQueryTableInserter {
Thread.currentThread().interrupt();
throw new IOException("Interrupted while inserting " + rowsToPublish);
} catch (ExecutionException e) {
- Throwables.propagate(e.getCause());
+ throw new RuntimeException(e.getCause());
}
if (!allErrors.isEmpty() && !backoff.atMaxAttempts()) {
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/454031af/sdks/java/core/src/test/java/org/apache/beam/sdk/util/ReduceFnTester.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/ReduceFnTester.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/ReduceFnTester.java
index f0d2a44..f296d65 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/ReduceFnTester.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/ReduceFnTester.java
@@ -58,7 +58,6 @@ import org.apache.beam.sdk.values.TupleTag;
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;
@@ -408,7 +407,7 @@ public class ReduceFnTester<InputT, OutputT, W extends BoundedWindow> {
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);
}
}
}));
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/454031af/sdks/java/core/src/test/java/org/apache/beam/sdk/util/TriggerTester.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/TriggerTester.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/TriggerTester.java
index 10d3d35..0889b4f 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/TriggerTester.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/TriggerTester.java
@@ -20,7 +20,6 @@ package org.apache.beam.sdk.util;
import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.base.Preconditions.checkState;
-
import static org.junit.Assert.assertTrue;
import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
@@ -45,7 +44,6 @@ import org.apache.beam.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;
@@ -257,7 +255,7 @@ public class TriggerTester<InputT, W extends BoundedWindow> {
windowedValues.add(WindowedValue.of(value, timestamp, assignedWindows, PaneInfo.NO_FIRING));
} catch (Exception e) {
- throw Throwables.propagate(e);
+ throw new RuntimeException(e);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/454031af/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java
index e605311..4f353df 100644
--- a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java
+++ b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java
@@ -44,7 +44,6 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Function;
import com.google.common.base.Joiner;
import com.google.common.base.Optional;
-import com.google.common.base.Throwables;
import com.google.common.collect.ComparisonChain;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
@@ -634,7 +633,7 @@ public class KafkaIO {
return new UnboundedKafkaReader<K, V>(
generateInitialSplits(1, options).get(0), checkpointMark);
} catch (Exception e) {
- Throwables.propagate(e);
+ throw new RuntimeException(e);
}
}
return new UnboundedKafkaReader<K, V>(this, checkpointMark);