[BEAM-3027] Correctly set output type on SourceId-stripper

mransley · Oct 9, 2017 · ca20e69 · ca20e69
1 parent b44056f
commit ca20e69
Showing 1 changed file with 2 additions and 2 deletions.
diff --git a/...flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingTransformTranslators.java b/...flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingTransformTranslators.java
@@ -188,9 +188,9 @@ public void translateNode(
  if (transform.getSource().requiresDeduping()) {
  source = nonDedupSource.keyBy(
  new ValueWithRecordIdKeySelector<T>())
- .transform("debuping", outputTypeInfo, new DedupingOperator<T>());
+ .transform("deduping", outputTypeInfo, new DedupingOperator<T>());
  } else {
- source = nonDedupSource.flatMap(new StripIdsMap<T>());
+ source = nonDedupSource.flatMap(new StripIdsMap<T>()).returns(outputTypeInfo);
  }
  } catch (Exception e) {
  throw new RuntimeException(