parameterize ThrowingConsumer by Exception type

apache · Feb 18, 2019 · 06a1fec · 06a1fec
1 parent db6559c
commit 06a1fec
Show file tree

Hide file tree

Showing 4 changed files with 8 additions and 7 deletions.
diff --git a/...n/src/main/java/org/apache/beam/runners/fnexecution/jobsubmission/InMemoryJobService.java b/...n/src/main/java/org/apache/beam/runners/fnexecution/jobsubmission/InMemoryJobService.java
@@ -74,7 +74,7 @@ public class InMemoryJobService extends JobServiceGrpc.JobServiceImplBase implem
   public static InMemoryJobService create(
       Endpoints.ApiServiceDescriptor stagingServiceDescriptor,
       Function<String, String> stagingServiceTokenProvider,
-      ThrowingConsumer<String> cleanupJobFn,
+      ThrowingConsumer<Exception, String> cleanupJobFn,
       JobInvoker invoker) {
     return new InMemoryJobService(
         stagingServiceDescriptor, stagingServiceTokenProvider, cleanupJobFn, invoker);
@@ -85,13 +85,13 @@ public static InMemoryJobService create(
   private final ConcurrentMap<String, String> stagingSessionTokens;
   private final Endpoints.ApiServiceDescriptor stagingServiceDescriptor;
   private final Function<String, String> stagingServiceTokenProvider;
-  private final ThrowingConsumer<String> cleanupJobFn;
+  private final ThrowingConsumer<Exception, String> cleanupJobFn;
   private final JobInvoker invoker;
 
   private InMemoryJobService(
       Endpoints.ApiServiceDescriptor stagingServiceDescriptor,
       Function<String, String> stagingServiceTokenProvider,
-      ThrowingConsumer<String> cleanupJobFn,
+      ThrowingConsumer<Exception, String> cleanupJobFn,
       JobInvoker invoker) {
     this.stagingServiceDescriptor = stagingServiceDescriptor;
     this.stagingServiceTokenProvider = stagingServiceTokenProvider;

diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/function/CloseableThrowingConsumer.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/function/CloseableThrowingConsumer.java
@@ -18,4 +18,5 @@
 package org.apache.beam.sdk.function;
 
 /** A {@link ThrowingConsumer} that can be closed. */
-public interface CloseableThrowingConsumer<T> extends AutoCloseable, ThrowingConsumer<T> {}
+public interface CloseableThrowingConsumer<ExceptionT extends Exception, T>
+    extends AutoCloseable, ThrowingConsumer<ExceptionT, T> {}
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/function/ThrowingConsumer.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/function/ThrowingConsumer.java
@@ -26,6 +26,6 @@
  * interfaces.
  */
 @FunctionalInterface
-public interface ThrowingConsumer<T> {
-  void accept(T t) throws Exception;
+public interface ThrowingConsumer<ExceptionT extends Exception, T> {
+  void accept(T t) throws ExceptionT;
 }
diff --git a/...va/harness/src/test/java/org/apache/beam/fn/harness/control/ProcessBundleHandlerTest.java b/...va/harness/src/test/java/org/apache/beam/fn/harness/control/ProcessBundleHandlerTest.java
@@ -75,7 +75,7 @@ public class ProcessBundleHandlerTest {
   @Rule public ExpectedException thrown = ExpectedException.none();
 
   @Mock private BeamFnDataClient beamFnDataClient;
-  @Captor private ArgumentCaptor<ThrowingConsumer<WindowedValue<String>>> consumerCaptor;
+  @Captor private ArgumentCaptor<ThrowingConsumer<Exception, WindowedValue<String>>> consumerCaptor;
 
   @Before
   public void setUp() {