Repository: apache/flink
File: flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/AbstractStreamTaskNetworkInput.java (Java)

Code before:
        return inputIndex;
    }

    @Override
    public CompletableFuture<?> getAvailableFuture() {
        if (currentRecordDeserializer != null) {
            return AVAILABLE;
        }
        return checkpointedInputGate.getAvailableFuture();
    }

    @Override
    public void close() throws IOException {
        // release the deserializers . this part should not ever fail
        Exception err = null;
        for (InputChannelInfo channelInfo : new ArrayList<>(recordDeserializers.keySet())) {
            final boolean hadError =
                    checkpointedInputGate.getChannel(channelInfo.getInputChannelIdx()).hasError();
            try {
                releaseDeserializer(channelInfo);
            } catch (Exception e) {
                if (hadError) {
                    LOG.warn(
                            "Ignoring deserializer release failure - the channel has encountered an error before");
                } else {
                    err = e;
                }
            }
        }
        if (err != null) {
            ExceptionUtils.rethrowIOException(err);
        }
    }

    protected void releaseDeserializer(InputChannelInfo channelInfo) {
        R deserializer = recordDeserializers.get(channelInfo);
        if (deserializer != null) {
            // recycle buffers and clear the deserializer.
            deserializer.clear();
            recordDeserializers.remove(channelInfo);
        }
    }
}


Review comment (suggestion):
!nit, If multiple channels throw, we will only capture the last one. Maybe it would make sense to chain them like: `err = ExceptionUtils.firstOrSuppressed(e, err);` or similar, WDYT

Code after:
    public int getInputIndex() {
        return inputIndex;
    }

    @Override
    public CompletableFuture<?> getAvailableFuture() {
        if (currentRecordDeserializer != null) {
            return AVAILABLE;
        }
        return checkpointedInputGate.getAvailableFuture();
    }

    @Override
    public void close() throws IOException {
        // WARNING: throwing an exception from this method might fail Task closing procedure and
        // terminate the TM
        Exception err = null;
        for (InputChannelInfo channelInfo : new ArrayList<>(recordDeserializers.keySet())) {
            final boolean hadError =
                    checkpointedInputGate.getChannel(channelInfo.getInputChannelIdx()).hasError();
            try {
                releaseDeserializer(channelInfo);
            } catch (Exception e) {
                if (hadError
                        && ExceptionUtils.findThrowable(e, RemoteTransportException.class)
                                .isPresent()) {
                    LOG.warn(
                            "Ignoring deserializer release failure - the channel {} has encountered a transport error before: {}",
                            channelInfo,
                            e.getMessage());
                } else {
                    err = ExceptionUtils.firstOrSuppressed(e, err);
                }
            }
        }
        if (err != null) {
            ExceptionUtils.rethrowIOException(err);
        }
    }

    protected void releaseDeserializer(InputChannelInfo channelInfo) {
        R deserializer = recordDeserializers.get(channelInfo);
        if (deserializer != null) {
            // recycle buffers and clear the deserializer.
            deserializer.clear();
            recordDeserializers.remove(channelInfo);
        }
    }
}