[
https://issues.apache.org/jira/browse/FLINK-14335?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Chesnay Schepler updated FLINK-14335:
-------------------------------------
Description:
The java version of the ExampleIntegrationTest is incorrect since it assumes
elements to arrive in the sink in order, but this isn't guaranteed since there
are 2 sink subtasks mutating a shared collection.
The scala example was modified correctly; it checks that elements are contained
without verifying the order.
{code}
public class ExampleIntegrationTest {
...
// configure your test environment
env.setParallelism(2);
...
// create a stream of custom elements and apply transformations
env.fromElements(1L, 21L, 22L)
.map(new IncrementMapFunction())
.addSink(new CollectSink());
// execute
env.execute();
// verify your results
assertEquals(Lists.newArrayList(2L, 42L, 44L), CollectSink.values);
}
// create a testing sink
private static class CollectSink implements SinkFunction<Long> {
// must be static
public static final List<Long> values = new ArrayList<>();
@Override
public synchronized void invoke(Long value) throws Exception {
values.add(value);
}
}
}
{code}
was:
The java version of the ExampleIntegrationTest is incorrect since it assumes
elements to arrive in the sink in order, but this isn't guaranteed since there
are 2 sink subtasks mutating a shared collection.
The scala example was modified correctly; it checks that elements are contained
with verifying the order.
{code}
public class ExampleIntegrationTest {
...
// configure your test environment
env.setParallelism(2);
...
// create a stream of custom elements and apply transformations
env.fromElements(1L, 21L, 22L)
.map(new IncrementMapFunction())
.addSink(new CollectSink());
// execute
env.execute();
// verify your results
assertEquals(Lists.newArrayList(2L, 42L, 44L), CollectSink.values);
}
// create a testing sink
private static class CollectSink implements SinkFunction<Long> {
// must be static
public static final List<Long> values = new ArrayList<>();
@Override
public synchronized void invoke(Long value) throws Exception {
values.add(value);
}
}
}
{code}
> Java version of ExampleIntegrationTest in testing docs is incorrect
> -------------------------------------------------------------------
>
> Key: FLINK-14335
> URL: https://issues.apache.org/jira/browse/FLINK-14335
> Project: Flink
> Issue Type: Task
> Components: Documentation, Tests
> Affects Versions: 1.9.0
> Reporter: Chesnay Schepler
> Priority: Major
> Fix For: 1.10.0, 1.9.2
>
>
> The java version of the ExampleIntegrationTest is incorrect since it assumes
> elements to arrive in the sink in order, but this isn't guaranteed since
> there are 2 sink subtasks mutating a shared collection.
> The scala example was modified correctly; it checks that elements are
> contained without verifying the order.
> {code}
> public class ExampleIntegrationTest {
> ...
> // configure your test environment
> env.setParallelism(2);
> ...
> // create a stream of custom elements and apply transformations
> env.fromElements(1L, 21L, 22L)
> .map(new IncrementMapFunction())
> .addSink(new CollectSink());
> // execute
> env.execute();
> // verify your results
> assertEquals(Lists.newArrayList(2L, 42L, 44L), CollectSink.values);
> }
> // create a testing sink
> private static class CollectSink implements SinkFunction<Long> {
> // must be static
> public static final List<Long> values = new ArrayList<>();
> @Override
> public synchronized void invoke(Long value) throws Exception {
> values.add(value);
> }
> }
> }
> {code}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
