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

Code before:
                currentRecordDeserializer != null,
                "currentRecordDeserializer has already been released");

        currentRecordDeserializer.setNextBuffer(bufferOrEvent.getBuffer());
    }

    protected R getActiveSerializer(InputChannelInfo channelInfo) {
        return recordDeserializers.get(channelInfo);
    }

    @Override
    public int getInputIndex() {
        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();

Review comment (bug):
🤔, this part shouldn't ever fail and we put try-catch right after. Should we update this comment explaining how can it fail?

Code after:
        checkState(
                currentRecordDeserializer != null,
                "currentRecordDeserializer has already been released");

        currentRecordDeserializer.setNextBuffer(bufferOrEvent.getBuffer());
    }

    protected R getActiveSerializer(InputChannelInfo channelInfo) {
        return recordDeserializers.get(channelInfo);
    }

    @Override
    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);
        }
    }