Put in another way, Flink seems to prematurely release slots which contain unconsumed data and, thus, eventually release a TM which then fails a consuming task. This might indicate that the remote task manager was lost.Īt .io.(PartitionRequestClientHandler.java:133)Īt .netty4.io.(AbstractChannelHandlerContext.java:237)Īfter skimming YarnFlinkResourceManager related code, it seems to me that Flink is releasing TMs when they’re idle, regardless of whether working TMs need them. io.: Connection unexpectedly closed by remote task manager ’the_previous_TM_used_by_read_task'. In the past, requesting input on a file would involve. This caused RemoteTransportException in the write task: Easily assign and track action items for more effective document collaboration.

However when the job was executing the write task, since only 4 TMs were needed, other 28 TMs were released. The read task worked perfectly with 32 TMs.

Our case is: Flink 1.5 batch mode, 32 parallelism to read data source and 4 parallelism to write data sink.

Remote task manager