rkhachatryan commented on a change in pull request #12120:
URL: https://github.com/apache/flink/pull/12120#discussion_r424919609
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriteRequest.java
##########
@@ -46,8 +48,27 @@ static CheckpointInProgressRequest completeOutput(long
checkpointId) {
return new CheckpointInProgressRequest("completeOutput",
checkpointId, ChannelStateCheckpointWriter::completeOutput, false);
}
- static ChannelStateWriteRequest write(long checkpointId,
InputChannelInfo info, Buffer... flinkBuffers) {
- return new CheckpointInProgressRequest("writeInput",
checkpointId, writer -> writer.writeInput(info, flinkBuffers),
recycle(flinkBuffers), false);
+ static ChannelStateWriteRequest write(long checkpointId,
InputChannelInfo info, CloseableIterator<Buffer> iterator) {
+ return new CheckpointInProgressRequest(
+ "writeInput",
+ checkpointId,
+ writer -> {
+ while (iterator.hasNext()) {
+ Buffer buffer = iterator.next();
+ checkArgument(buffer.isBuffer());
+ writer.writeInput(info, buffer);
+ }
+ },
+ throwable -> {
+ try {
+ iterator.close();
+ } catch (Exception e) {
+ final RuntimeException re = new
RuntimeException("unable to close iterator", e);
+ re.addSuppressed(throwable);
Review comment:
~~It depends on which exception is thrown afterward. Currently, it is
`re` to which `throwable` is added as suppressed.~~
~~Do you think we should throw `throwable` instead?~~
Irrelevant after addressing the next comment.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
