[FLINK-18050][task][checkpointing] Simplify ChannelStateCheckpointWri…

…ter interface

flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateCheckpointWriter.java

@@ -108,32 +108,30 @@ class ChannelStateCheckpointWriter {
  runWithChecks(() -> serializer.writeHeader(dataStream));
  }
 
- void writeInput(InputChannelInfo info, Buffer... flinkBuffers) throws Exception {
- write(inputChannelOffsets, info, flinkBuffers, !allInputsReceived);
+ void writeInput(InputChannelInfo info, Buffer buffer) throws Exception {
+ write(inputChannelOffsets, info, buffer, !allInputsReceived);
  }
 
- void writeOutput(ResultSubpartitionInfo info, Buffer... flinkBuffers) throws Exception {
- write(resultSubpartitionOffsets, info, flinkBuffers, !allOutputsReceived);
+ void writeOutput(ResultSubpartitionInfo info, Buffer buffer) throws Exception {
+ write(resultSubpartitionOffsets, info, buffer, !allOutputsReceived);
  }
 
- private <K> void write(Map<K, StateContentMetaInfo> offsets, K key, Buffer[] flinkBuffers, boolean precondition) throws Exception {
+ private <K> void write(Map<K, StateContentMetaInfo> offsets, K key, Buffer buffer, boolean precondition) throws Exception {
  try {
  if (result.isDone()) {
  return;
  }
  runWithChecks(() -> {
  checkState(precondition);
  long offset = checkpointStream.getPos();
- serializer.writeData(dataStream, flinkBuffers);
+ serializer.writeData(dataStream, buffer);
  long size = checkpointStream.getPos() - offset;
  offsets
  .computeIfAbsent(key, unused -> new StateContentMetaInfo())
  .withDataAdded(offset, size);
  });
  } finally {
- for (Buffer flinkBuffer : flinkBuffers) {
- flinkBuffer.recycleBuffer();
- }
+ buffer.recycleBuffer();
  }
  }