[ https://issues.apache.org/jira/browse/FLINK-8268?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16308136#comment-16308136 ]
ASF GitHub Bot commented on FLINK-8268: --------------------------------------- Github user pnowojski commented on a diff in the pull request: https://github.com/apache/flink/pull/5193#discussion_r159235730 --- 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 -- It wouldn't suffice, because rich function's `MockEnvironment` must be closed. It would have to be something like: ``` Optional<MockEnvironment> context = Optional.empty(); try { if (sourceFunction instanceof RichFunction) { context = setupRichSourceFunction(sourceFunction); } return runSourceFunction(sourceFunction); } finally { context.ifPresent(MockEnvironment::close); } ``` Which in my opinion is uglier :( > 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)