title: org.apache.flink.runtime.io.network.netty.exception.RemoteTransportException: Lost connection to task manager
body: <p>I'm working on a <strong>project that uses flink</strong> (<strong>Version - 1.4.2</strong>) for bulk data ingestion to <strong>my graph database</strong> (<strong>Janusgraph</strong>). Data ingestion has two phases, one is vertex data ingestion and the other is edge data ingestion to graph db. <strong>Vertex data ingestion runs without any issue but during edge ingestion</strong> i'm getting an error saying <strong>Lost connection to task manager taskmanagerName</strong>. The detailed error traceback from <code>flink-taskmanager-b6f46f6c8-fgtlw</code> is attached below:</p>
  
  <pre><code>2019-08-01 18:13:26,025 ERROR org.apache.flink.runtime.operators.BatchTask 
    - Error in task code:  CHAIN Join(Remap EDGES id: TO) -&gt; Map (Key Extractor) -&gt; Combine (Deduplicate edges including bi-directional edges) (62/80)
  org.apache.flink.runtime.io.network.netty.exception.RemoteTransportException: Lost connection to task manager 'flink-taskmanager-b6f46f6c8-gcxnm/10.xx.xx.xx:6121'. 
  This indicates that the remote task manager was lost.
  at org.apache.flink.runtime.io.network.netty.PartitionRequestClientHandler.exceptionCaught(PartitionRequestClientHandler.java:146)
  at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeExceptionCaught(AbstractChannelHandlerContext.java:275)
  at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireExceptionCaught(AbstractChannelHandlerContext.java:253)
  at org.apache.flink.shaded.netty4.io.netty.channel.ChannelInboundHandlerAdapter.exceptionCaught(ChannelInboundHandlerAdapter.java:131)
  at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeExceptionCaught(AbstractChannelHandlerContext.java:275)
  at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireExceptionCaught(AbstractChannelHandlerContext.java:253)
  at org.apache.flink.shaded.netty4.io.netty.channel.ChannelInboundHandlerAdapter.exceptionCaught(ChannelInboundHandlerAdapter.java:131)
  at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeExceptionCaught(AbstractChannelHandlerContext.java:275)
  at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireExceptionCaught(AbstractChannelHandlerContext.java:253)
  at org.apache.flink.shaded.netty4.io.netty.channel.ChannelHandlerAdapter.exceptionCaught(ChannelHandlerAdapter.java:79)
  at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeExceptionCaught(AbstractChannelHandlerContext.java:275)
  at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireExceptionCaught(AbstractChannelHandlerContext.java:253)
  at org.apache.flink.shaded.netty4.io.netty.channel.DefaultChannelPipeline.fireExceptionCaught(DefaultChannelPipeline.java:835)
  at org.apache.flink.shaded.netty4.io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.handleReadException(AbstractNioByteChannel.java:87)
  at org.apache.flink.shaded.netty4.io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:162)
  at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:511)
  at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:468)
  at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:382)
  at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:354)
  at org.apache.flink.shaded.netty4.io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:111)
  at java.lang.Thread.run(Thread.java:748)
  
  Caused by: java.io.IOException: Connection reset by peer
  at sun.nio.ch.FileDispatcherImpl.read0(Native Method)
  at sun.nio.ch.SocketDispatcher.read(SocketDispatcher.java:39)
  at sun.nio.ch.IOUtil.readIntoNativeBuffer(IOUtil.java:223)
  at sun.nio.ch.IOUtil.read(IOUtil.java:192)
  at sun.nio.ch.SocketChannelImpl.read(SocketChannelImpl.java:380)
  at org.apache.flink.shaded.netty4.io.netty.buffer.PooledUnsafeDirectByteBuf.setBytes(PooledUnsafeDirectByteBuf.java:311)
  at org.apache.flink.shaded.netty4.io.netty.buffer.AbstractByteBuf.writeBytes(AbstractByteBuf.java:881)
  at org.apache.flink.shaded.netty4.io.netty.channel.socket.nio.NioSocketChannel.doReadBytes(NioSocketChannel.java:241)
  at org.apache.flink.shaded.netty4.io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:119)
  ... 6 more
  </code></pre>
  
  <p>For ease of understanding lets say:</p>
  
  <p><code>flink-taskmanager-b6f46f6c8-gcxnm</code> as TM1 and </p>
  
  <p><code>flink-taskmanager-b6f46f6c8-fgtlw</code> as TM2</p>
  
  <p>On debugging I was able to find that the <strong>TM1 requested for <code>ResultPartition (RPP)</code> from  TM2 and the TM2 started to send ResultPartition to TM1</strong>. But on checking the logs from <strong>TM1</strong> we found that it <strong>waited for long time to get RP from TM2</strong> but <strong>after some time it started to deregister the accepted task</strong>. We believe deregistring task after netty remote transport exception caused TM2 to send <code>Lost Taskmanager</code> error for the specific job.
  Both <strong>taskmanagers are running in separate ec2 instance (m4.2xlarge)</strong>. I have verified <strong>cpu and memory utilization of both instances</strong> and was able to see <strong>all metric within limit</strong>.</p>
  
  <p>Can you please tell me why taskmanager is acting weird like this and a way to fix this issue.</p>
  
  <p>Thanks in advance</p>