[
https://issues.apache.org/jira/browse/FLINK-8268?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16300688#comment-16300688
]
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_r158394094
--- Diff:
flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/TwoPhaseCommitSinkFunctionTest.java
---
@@ -277,88 +263,67 @@ public void
testLogTimeoutAlmostReachedWarningDuringRecovery() throws Exception
private void assertExactlyOnce(List<String> expectedValues) throws
IOException {
ArrayList<String> actualValues = new ArrayList<>();
- for (File file : targetDirectory.listFiles()) {
- actualValues.addAll(Files.readAllLines(file.toPath(),
Charset.defaultCharset()));
+ for (String name : targetDirectory.listFiles()) {
+ actualValues.addAll(targetDirectory.read(name));
}
Collections.sort(actualValues);
Collections.sort(expectedValues);
assertEquals(expectedValues, actualValues);
}
- private class FileBasedSinkFunction extends
TwoPhaseCommitSinkFunction<String, FileTransaction, Void> {
+ private class ContentDumpSinkFunction extends
TwoPhaseCommitSinkFunction<String, ContentTransaction, Void> {
- public FileBasedSinkFunction() {
+ public ContentDumpSinkFunction() {
super(
- new KryoSerializer<>(FileTransaction.class, new
ExecutionConfig()),
+ new KryoSerializer<>(ContentTransaction.class,
new ExecutionConfig()),
VoidSerializer.INSTANCE, clock);
-
- if (!tmpDirectory.isDirectory() ||
!targetDirectory.isDirectory()) {
- throw new IllegalArgumentException();
- }
}
@Override
- protected void invoke(FileTransaction transaction, String
value, Context context) throws Exception {
- transaction.writer.write(value);
+ protected void invoke(ContentTransaction transaction, String
value, Context context) throws Exception {
+ transaction.tmpContentWriter.write(value);
}
@Override
- protected FileTransaction beginTransaction() throws Exception {
- File tmpFile = new File(tmpDirectory,
UUID.randomUUID().toString());
- return new FileTransaction(tmpFile);
+ protected ContentTransaction beginTransaction() throws
Exception {
+ return new
ContentTransaction(tmpDirectory.createWriter(UUID.randomUUID().toString()));
}
@Override
- protected void preCommit(FileTransaction transaction) throws
Exception {
- transaction.writer.flush();
- transaction.writer.close();
+ protected void preCommit(ContentTransaction transaction) throws
Exception {
+ transaction.tmpContentWriter.flush();
--- End diff --
Should also call `transaction.tmpContentWriter.close()` after the precommit?
> 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)
