[
https://issues.apache.org/jira/browse/FLINK-8268?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16300726#comment-16300726
]
ASF GitHub Bot commented on FLINK-8268:
---------------------------------------
Github user tzulitai commented on a diff in the pull request:
https://github.com/apache/flink/pull/5193#discussion_r158401540
--- Diff:
flink-streaming-java/src/test/java/org/apache/flink/streaming/util/SourceFunctionUtil.java
---
@@ -42,22 +41,32 @@
public class SourceFunctionUtil {
public static <T extends Serializable> List<T>
runSourceFunction(SourceFunction<T> sourceFunction) throws Exception {
- final List<T> outputs = new ArrayList<T>();
-
if (sourceFunction instanceof RichFunction) {
+ return runRichSourceFunction(sourceFunction);
+ }
+ else {
+ return runNonRichSourceFunction(sourceFunction);
+ }
+ }
+ private static <T extends Serializable> List<T>
runRichSourceFunction(SourceFunction<T> sourceFunction) throws Exception {
+ try (MockEnvironment environment = new
MockEnvironment("MockTask", 3 * 1024 * 1024, new MockInputSplitProvider(),
1024)) {
AbstractStreamOperator<?> operator =
mock(AbstractStreamOperator.class);
when(operator.getExecutionConfig()).thenReturn(new
ExecutionConfig());
- RuntimeContext runtimeContext = new
StreamingRuntimeContext(
- operator,
- new MockEnvironment("MockTask", 3 *
1024 * 1024, new MockInputSplitProvider(), 1024),
- new HashMap<String, Accumulator<?,
?>>());
-
+ RuntimeContext runtimeContext = new
StreamingRuntimeContext(
+ operator,
+ environment,
+ new HashMap<>());
((RichFunction)
sourceFunction).setRuntimeContext(runtimeContext);
-
((RichFunction) sourceFunction).open(new
Configuration());
+
+ return runNonRichSourceFunction(sourceFunction);
--- End diff --
The fact that `runRichSourceFunction` ended with a
`runNonRichSourceFunction` was a bit surprising for me.
It maybe can be read better if we simply had this:
```
if (sourceFunction instanceof RichFunction) {
setupRichSourceFunction(sourceFunction);
}
return runSourceFunction(sourceFunction);
```
where `setupUpSourceFunction` does all the operator / env / context mocking
and invoking `setRuntimeContext`, `open` on the source function.
What do you think?
> Test instability for 'TwoPhaseCommitSinkFunctionTest'
> -----------------------------------------------------
>
> Key: FLINK-8268
> URL: https://issues.apache.org/jira/browse/FLINK-8268
> Project: Flink
> Issue Type: Bug
> Components: Tests
> Affects Versions: 1.5.0
> Reporter: Stephan Ewen
> Assignee: Piotr Nowojski
> Priority: Critical
> Labels: test-stability
>
> The following exception / failure message occurs.
> {code}
> Tests run: 5, Failures: 0, Errors: 1, Skipped: 0, Time elapsed: 1.824 sec <<<
> FAILURE! - in
> org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunctionTest
> testIgnoreCommitExceptionDuringRecovery(org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunctionTest)
> Time elapsed: 0.068 sec <<< ERROR!
> java.lang.Exception: Could not complete snapshot 0 for operator MockTask
> (1/1).
> at java.io.FileOutputStream.writeBytes(Native Method)
> at java.io.FileOutputStream.write(FileOutputStream.java:326)
> at sun.nio.cs.StreamEncoder.writeBytes(StreamEncoder.java:221)
> at sun.nio.cs.StreamEncoder.implFlushBuffer(StreamEncoder.java:291)
> at sun.nio.cs.StreamEncoder.implFlush(StreamEncoder.java:295)
> at sun.nio.cs.StreamEncoder.flush(StreamEncoder.java:141)
> at java.io.OutputStreamWriter.flush(OutputStreamWriter.java:229)
> at java.io.BufferedWriter.flush(BufferedWriter.java:254)
> at
> org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunctionTest$FileBasedSinkFunction.preCommit(TwoPhaseCommitSinkFunctionTest.java:313)
> at
> org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunctionTest$FileBasedSinkFunction.preCommit(TwoPhaseCommitSinkFunctionTest.java:288)
> at
> org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunction.snapshotState(TwoPhaseCommitSinkFunction.java:290)
> at
> org.apache.flink.streaming.util.functions.StreamingFunctionUtils.trySnapshotFunctionState(StreamingFunctionUtils.java:118)
> at
> org.apache.flink.streaming.util.functions.StreamingFunctionUtils.snapshotFunctionState(StreamingFunctionUtils.java:99)
> at
> org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.snapshotState(AbstractUdfStreamOperator.java:90)
> at
> org.apache.flink.streaming.api.operators.AbstractStreamOperator.snapshotState(AbstractStreamOperator.java:357)
> at
> org.apache.flink.streaming.util.AbstractStreamOperatorTestHarness.snapshot(AbstractStreamOperatorTestHarness.java:459)
> at
> org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunctionTest.testIgnoreCommitExceptionDuringRecovery(TwoPhaseCommitSinkFunctionTest.java:208)
> {code}
--
This message was sent by Atlassian JIRA
(v6.4.14#64029)
