You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "John Lonergan (Jira)" <ji...@apache.org> on 2019/08/19 15:47:00 UTC
[jira] [Commented] (FLINK-10382) Writer has already been opened
while using AvroKeyValueSinkWriter and BucketingSink
[ https://issues.apache.org/jira/browse/FLINK-10382?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16910497#comment-16910497 ]
John Lonergan commented on FLINK-10382:
---------------------------------------
Just fell over this same problem - Fundamentally this class doesn't work as far as I can tell.
Presumably there are missing tests - which raises a question around QA in Flink - what is the overall quality of testing in Flink?
I suggest that rather than just documenting this class as deprecated it comes with a comment saying it's broken and should be avoided - or just delete it entirely if this class has been abandoned.
---
As far as a "fix" I think something like this ought to work. Flush needed first to avoid potential data loss.
{{@Override public void close() throws IOException {
if (keyValueWriter != null) {
flush();
super.close();
keyValueWriter = null;
} else { // need to make sure we close this if we never created the Key/Value Writer.
super.close();
}
}
}}
Is the bucketing sink abandoned?
This close issue only seems to be safe on StringWriter and SequenceFileWriter, the first because there's nothing to close and the second because it knows whether it owns the stream or not.
> Writer has already been opened while using AvroKeyValueSinkWriter and BucketingSink
> -----------------------------------------------------------------------------------
>
> Key: FLINK-10382
> URL: https://issues.apache.org/jira/browse/FLINK-10382
> Project: Flink
> Issue Type: Bug
> Components: Connectors / FileSystem
> Affects Versions: 1.5.0, 1.6.0
> Reporter: Chengzhi Zhao
> Priority: Major
>
> I am using flink 1.6.0 and I am using AvroKeyValueSinkWriter and BucketingSink to S3.
>
> After the application running for a while ~ 20 mins, I got an *exception: java.lang.IllegalStateException: Writer has already been opened*
> {code:java}
> 2018-09-17 15:40:23,771 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Triggering checkpoint 4 @ 1537198823640 for job 8f9ab122fb7452714465eb1e1989e4d7.
> 2018-09-17 15:41:27,805 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Sink: Unnamed (2/16) (25914cb3f77c8e4271b0fb6ea597ed50) switched from RUNNING to FAILED.
> java.lang.IllegalStateException: Writer has already been opened
> at org.apache.flink.streaming.connectors.fs.StreamWriterBase.open(StreamWriterBase.java:68)
> at org.apache.flink.streaming.connectors.fs.AvroKeyValueSinkWriter.open(AvroKeyValueSinkWriter.java:150)
> at org.apache.flink.streaming.connectors.fs.bucketing.BucketingSink.openNewPartFile(BucketingSink.java:583)
> at org.apache.flink.streaming.connectors.fs.bucketing.BucketingSink.invoke(BucketingSink.java:458)
> at org.apache.flink.streaming.api.functions.sink.SinkFunction.invoke(SinkFunction.java:52)
> at org.apache.flink.streaming.api.operators.StreamSink.processElement(StreamSink.java:56)
> at org.apache.flink.streaming.runtime.io.StreamInputProcessor.processInput(StreamInputProcessor.java:202)
> at org.apache.flink.streaming.runtime.tasks.OneInputStreamTask.run(OneInputStreamTask.java:104)
> at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:306)
> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:703)
> at java.lang.Thread.run(Thread.java:748)
> 2018-09-17 15:41:27,808 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Job Stream to Stream Join (8f9ab122fb7452714465eb1e1989e4d7) switched from state RUNNING to FAILING.
> java.lang.IllegalStateException: Writer has already been opened
> at org.apache.flink.streaming.connectors.fs.StreamWriterBase.open(StreamWriterBase.java:68)
> at org.apache.flink.streaming.connectors.fs.AvroKeyValueSinkWriter.open(AvroKeyValueSinkWriter.java:150)
> at org.apache.flink.streaming.connectors.fs.bucketing.BucketingSink.openNewPartFile(BucketingSink.java:583)
> at org.apache.flink.streaming.connectors.fs.bucketing.BucketingSink.invoke(BucketingSink.java:458)
> at org.apache.flink.streaming.api.functions.sink.SinkFunction.invoke(SinkFunction.java:52)
> at org.apache.flink.streaming.api.operators.StreamSink.processElement(StreamSink.java:56)
> at org.apache.flink.streaming.runtime.io.StreamInputProcessor.processInput(StreamInputProcessor.java:202)
> at org.apache.flink.streaming.runtime.tasks.OneInputStreamTask.run(OneInputStreamTask.java:104)
> at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:306)
> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:703)
> at java.lang.Thread.run(Thread.java:748)
> {code}
> After checking the code, I think the issue might be related to AvroKeyValueSinkWriter.java and led to the writer has not been closed completely. I also noticed this change and affect 1.5+
> [https://github.com/apache/flink/commit/915213c7afaf3f9d04c240f43d88710280d844e3#diff-86c35c993fdb0c482544951b376e5ea6]
> I created my own AvroKeyValueSinkWriter class and implement the code similar as v1.4, it seems running fine now.
> {code:java}
> @Override
> public void close() throws IOException {
> try {
> super.close();
> } finally {
> if (keyValueWriter != null) {
> keyValueWriter.close();
> }
> }
> }
> {code}
> I am curious if anyone had the similar issue, Appreciate anyone has insights on it. Thanks!
>
--
This message was sent by Atlassian Jira
(v8.3.2#803003)