Uploaded image for project: 'Flink'
  1. Flink
  2. FLINK-18832

BoundedBlockingSubpartition does not work with StreamTask

    XMLWordPrintableJSON

    Details

      Description

      The BoundedBlockingSubpartition does not work with a StreamTask because the StreamTask instantiates an OutputFlusher which concurrently accesses the BoundedBlockingSubpartition. This concurrency can lead to a double closing of the underlying BufferConsumer which manifests in this stack trace:

      [9:15 PM] Caused by: org.apache.flink.shaded.netty4.io.netty.util.IllegalReferenceCountException: refCnt: 0, increment: 1
      	at org.apache.flink.shaded.netty4.io.netty.util.internal.ReferenceCountUpdater.retain0(ReferenceCountUpdater.java:123)
      	at org.apache.flink.shaded.netty4.io.netty.util.internal.ReferenceCountUpdater.retain(ReferenceCountUpdater.java:110)
      	at org.apache.flink.shaded.netty4.io.netty.buffer.AbstractReferenceCountedByteBuf.retain(AbstractReferenceCountedByteBuf.java:80)
      	at org.apache.flink.runtime.io.network.buffer.NetworkBuffer.retainBuffer(NetworkBuffer.java:174)
      	at org.apache.flink.runtime.io.network.buffer.NetworkBuffer.retainBuffer(NetworkBuffer.java:47)
      	at org.apache.flink.runtime.io.network.buffer.ReadOnlySlicedNetworkBuffer.retainBuffer(ReadOnlySlicedNetworkBuffer.java:127)
      	at org.apache.flink.runtime.io.network.buffer.ReadOnlySlicedNetworkBuffer.retainBuffer(ReadOnlySlicedNetworkBuffer.java:41)
      	at org.apache.flink.runtime.io.network.buffer.BufferConsumer.build(BufferConsumer.java:108)
      	at org.apache.flink.runtime.io.network.partition.BoundedBlockingSubpartition.writeAndCloseBufferConsumer(BoundedBlockingSubpartition.java:156)
      	at org.apache.flink.runtime.io.network.partition.BoundedBlockingSubpartition.flushCurrentBuffer(BoundedBlockingSubpartition.java:144)
      	at org.apache.flink.runtime.io.network.partition.BoundedBlockingSubpartition.flush(BoundedBlockingSubpartition.java:135)
      	at org.apache.flink.runtime.io.network.partition.ResultPartition.flushAll(ResultPartition.java:245)
      	at org.apache.flink.runtime.io.network.api.writer.RecordWriter.flushAll(RecordWriter.java:183)
      	at org.apache.flink.streaming.runtime.io.RecordWriterOutput.flush(RecordWriterOutput.java:156)
      	at org.apache.flink.streaming.runtime.tasks.OperatorChain.flushOutputs(OperatorChain.java:344)
      	at org.apache.flink.streaming.runtime.tasks.StreamTask.afterInvoke(StreamTask.java:602)
      	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:544)
      	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:721)
      	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:546)
      	at java.lang.Thread.run(Thread.java:748)
      

        Attachments

          Issue Links

            Activity

              People

              • Assignee:
                zjwang Zhijiang
                Reporter:
                trohrmann Till Rohrmann
              • Votes:
                0 Vote for this issue
                Watchers:
                11 Start watching this issue

                Dates

                • Created:
                  Updated:
                  Resolved: