Github user zhijiangW commented on a diff in the pull request: https://github.com/apache/flink/pull/4533#discussion_r146187454 --- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedClientHandler.java --- @@ -269,4 +315,49 @@ private void decodeBufferOrEvent(RemoteInputChannel inputChannel, NettyMessage.B bufferOrEvent.releaseBuffer(); } } + + private void writeAndFlushNextMessageIfPossible(Channel channel) { + if (channelError.get() != null || !channel.isWritable()) { + return; + } + + while (true) { + RemoteInputChannel inputChannel = inputChannelsWithCredit.poll(); + + // The input channel may be null because of the write callbacks that are executed + // after each write, and it is also no need to notify credit for released channel. + if (inputChannel == null || inputChannel.isReleased()) { + return; + } + + AddCredit msg = new AddCredit( + inputChannel.getPartitionId(), + inputChannel.getAndResetCredit(), + inputChannel.getInputChannelId()); + + // Write and flush and wait until this is done before + // trying to continue with the next input channel. + channel.writeAndFlush(msg).addListener(writeListener); + + return; --- End diff -- Sorry, I misunderstood your meaning before. This loop indeed has some redundancies. I will improve it together with UT fixing.
---