dmvk commented on a change in pull request #13353:
URL: https://github.com/apache/beam/pull/13353#discussion_r526640194
##########
File path:
runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkStreamingPipelineTranslatorTest.java
##########
@@ -120,4 +138,95 @@ public void testAutoBalanceShardKeyCacheMaxSize() throws
Exception {
assertThat(
fn.getCache().size(),
equalTo(FlinkAutoBalancedShardKeyShardingFunction.CACHE_MAX_SIZE));
}
+
+ @Test
+ public void testStatefulParDoAfterCombineChaining() {
+ final JobGraph stablePartitioning =
getStatefulParDoAfterCombineChainingJobGraph(true);
+ final JobGraph unstablePartitioning =
getStatefulParDoAfterCombineChainingJobGraph(false);
+ // We expect an extra shuffle stage for unstable partitioning.
+ Assert.assertEquals(
+ 1,
+ Iterables.size(unstablePartitioning.getVertices())
+ - Iterables.size(stablePartitioning.getVertices()));
+ }
+
+ private JobGraph getStatefulParDoAfterCombineChainingJobGraph(boolean
stablePartitioning) {
+ final StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+ final FlinkStreamingPipelineTranslator translator =
+ new FlinkStreamingPipelineTranslator(env,
PipelineOptionsFactory.create());
+ final PipelineOptions pipelineOptions = PipelineOptionsFactory.create();
+ pipelineOptions.setRunner(FlinkRunner.class);
Review comment:
unfortunately yes, otherwise pipeline.create() would fail
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
