[flink] 02/11: [hotfix][network] ensure deserialization buffer capacity for the whole record length

[nkruber]

This is an automated email from the ASF dual-hosted git repository.

nkruber pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git

commit bbb8b0a10f4f48c47ffbeb325c7e3ce0203a552f
Author: Nico Kruber <n...@data-artisans.com>
AuthorDate: Thu Sep 13 12:14:05 2018 +0200

    [hotfix][network] ensure deserialization buffer capacity for the whole 
record length
    
    Once we know the record length and if we are not spilling, we should size 
the
    buffer immediately to the expected record size, and not incrementally for 
each
    received buffer chunk.
---
 .../serialization/SpillingAdaptiveSpanningRecordDeserializer.java | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git 
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/SpillingAdaptiveSpanningRecordDeserializer.java
 
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/SpillingAdaptiveSpanningRecordDeserializer.java
index 41ee03d..196287b 100644
--- 
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/SpillingAdaptiveSpanningRecordDeserializer.java
+++ 
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/SpillingAdaptiveSpanningRecordDeserializer.java
@@ -485,7 +485,7 @@ public class SpillingAdaptiveSpanningRecordDeserializer<T 
extends IOReadableWrit
                        }
                        else {
                                // collect in memory
-                               ensureBufferCapacity(numBytesChunk);
+                               ensureBufferCapacity(nextRecordLength);
                                partial.segment.get(partial.position, buffer, 
0, numBytesChunk);
                        }
 
@@ -515,6 +515,8 @@ public class SpillingAdaptiveSpanningRecordDeserializer<T 
extends IOReadableWrit
                                        segmentRemaining -= toPut;
                                        if (this.recordLength > 
THRESHOLD_FOR_SPILLING) {
                                                this.spillingChannel = 
createSpillingChannel();
+                                       } else {
+                                               
ensureBufferCapacity(this.recordLength);
                                        }
                                }
                        }
@@ -527,9 +529,7 @@ public class SpillingAdaptiveSpanningRecordDeserializer<T 
extends IOReadableWrit
                                // spill to file
                                ByteBuffer toWrite = 
segment.wrap(segmentPosition, toCopy);
                                this.spillingChannel.write(toWrite);
-                       }
-                       else {
-                               ensureBufferCapacity(accumulatedRecordBytes + 
toCopy);
+                       } else {
                                segment.get(segmentPosition, buffer, 
this.accumulatedRecordBytes, toCopy);
                        }