rkhachatryan commented on a change in pull request #12120:
URL: https://github.com/apache/flink/pull/12120#discussion_r424938864
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/SpanningWrapper.java
##########
@@ -183,22 +196,26 @@ private void updateLength(int length) throws IOException {
@Override
public CloseableIterator<Buffer> getUnconsumedSegment() throws
IOException {
if (isReadingLength()) {
- return singleBufferIterator(copyLengthBuffer());
+ return
singleBufferIterator(wrapCopy(lengthBuffer.array(), 0,
lengthBuffer.position()));
} else if (isAboveSpillingThreshold()) {
- throw new UnsupportedOperationException("Unaligned
checkpoint currently do not support spilled records.");
+ return createSpilledDataIterator();
} else if (recordLength == -1) {
- return CloseableIterator.empty(); // no remaining
partial length or data
+ return empty(); // no remaining partial length or data
} else {
return singleBufferIterator(copyDataBuffer());
}
}
- private MemorySegment copyLengthBuffer() {
- int position = lengthBuffer.position();
- MemorySegment segment =
MemorySegmentFactory.allocateUnpooledSegment(position);
- lengthBuffer.position(0);
- segment.put(0, lengthBuffer, position);
- return segment;
+ @SuppressWarnings("unchecked")
+ private CloseableIterator<Buffer> createSpilledDataIterator() throws
IOException {
+ if (spillingChannel != null) {
+ spillingChannel.force(false);
+ }
+ return CloseableIterator.flatten(
+ toSingleBufferIterator(wrapInt(recordLength)),
+ new FileBasedBufferIterator(spillFile,
min(accumulatedRecordBytes, recordLength), fileBufferSize),
+ leftOverData == null ? empty() :
toSingleBufferIterator(wrapCopy(leftOverData.getArray(), leftOverStart,
leftOverLimit))
+ );
Review comment:
It flattens 3 iterators, one of which can be empty. I tried to put it
all into `FileBasedBufferIterator` but it doesn't simplify anything (as there
will be many checks there) and makes it more error-prone.
----------------------------------------------------------------
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]
