Merge pull request apache#12 from bsidhom/output-receiver-empty

Only access output coder if available
aljoscha · Mar 13, 2018 · 6bbd8a3 · 6bbd8a3
2 parents 8a2d352 + 9714449
commit 6bbd8a3
Showing 1 changed file with 21 additions and 20 deletions.
diff --git a/...ava/org/apache/beam/runners/flink/translation/functions/FlinkExecutableStageFunction.java b/...ava/org/apache/beam/runners/flink/translation/functions/FlinkExecutableStageFunction.java
@@ -111,29 +111,30 @@ public void mapPartition(Iterable<WindowedValue<InputT>> input,
     Map<BeamFnApi.Target, Coder<WindowedValue<?>>> outputCoders =
         processBundleDescriptor.getOutputTargetCoders();
     BeamFnApi.Target outputTarget = null;
+    SdkHarnessClient.RemoteOutputReceiver<WindowedValue<OutputT>> mainOutputReceiver = null;
     if (outputCoders.size() > 0) {
       outputTarget = Iterables.getOnlyElement(outputCoders.keySet());
-    }
-    Coder<?> outputCoder = Iterables.getOnlyElement(outputCoders.values());
-    SdkHarnessClient.RemoteOutputReceiver<WindowedValue<OutputT>> mainOutputReceiver =
-        new SdkHarnessClient.RemoteOutputReceiver<WindowedValue<OutputT>>() {
-          @Override
-          public Coder<WindowedValue<OutputT>> getCoder() {
-            @SuppressWarnings("unchecked")
-            Coder<WindowedValue<OutputT>> result = (Coder<WindowedValue<OutputT>>) outputCoder;
-            return result;
-          }
+      Coder<?> outputCoder = Iterables.getOnlyElement(outputCoders.values());
+      mainOutputReceiver =
+          new SdkHarnessClient.RemoteOutputReceiver<WindowedValue<OutputT>>() {
+            @Override
+            public Coder<WindowedValue<OutputT>> getCoder() {
+              @SuppressWarnings("unchecked")
+              Coder<WindowedValue<OutputT>> result = (Coder<WindowedValue<OutputT>>) outputCoder;
+              return result;
+            }
 
-          @Override
-          public FnDataReceiver<WindowedValue<OutputT>> getReceiver() {
-            return new FnDataReceiver<WindowedValue<OutputT>>() {
-              @Override
-              public void accept(WindowedValue<OutputT> input) throws Exception {
-                collector.collect(input);
-              }
-            };
-          }
-        };
+            @Override
+            public FnDataReceiver<WindowedValue<OutputT>> getReceiver() {
+              return new FnDataReceiver<WindowedValue<OutputT>>() {
+                @Override
+                public void accept(WindowedValue<OutputT> input) throws Exception {
+                  collector.collect(input);
+                }
+              };
+            }
+          };
+    }
     Map<BeamFnApi.Target,
         SdkHarnessClient.RemoteOutputReceiver<?>> receiverMap;
     if (outputTarget == null) {