[FLINK-19023][network] Remove unnecessary buffer pruning in RecordSer…

…ializer

This removes the behavior of the SpanningRecordSerializer to prune its internal serialization buffer
under special circumstances.

Previously, the buffer was pruned when:
  - The buffer becomes larger than a certain threshold (5MB)
  - The full record end lines up exactly with a full buffer length (this change got introduced
    at some point, it is not clear what the purpose is)

This optimization virtually never kicks in (because of the second condition) and also is unnecessary.
There is only a single serializer on the sender side, so this will not help to reduce the maximum memory
footprint needed in any way.

flink-core/src/main/java/org/apache/flink/core/memory/DataOutputSerializer.java

@@ -20,9 +20,6 @@
 
 import org.apache.flink.util.Preconditions;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import java.io.EOFException;
 import java.io.IOException;
 import java.io.UTFDataFormatException;
@@ -35,14 +32,6 @@
  */
 public class DataOutputSerializer implements DataOutputView, MemorySegmentWritable {
 
-	private static final Logger LOG = LoggerFactory.getLogger(DataOutputSerializer.class);
-
-	private static final int PRUNE_BUFFER_THRESHOLD = 5 * 1024 * 1024;
-
-	// ------------------------------------------------------------------------
-
-	private final byte[] startBuffer;
-
 	private byte[] buffer;
 
 	private int position;
@@ -56,8 +45,7 @@ public DataOutputSerializer(int startSize) {
 			throw new IllegalArgumentException();
 		}
 
-		this.startBuffer = new byte[startSize];
-		this.buffer = this.startBuffer;
+		this.buffer = new byte[startSize];
 		this.wrapper = ByteBuffer.wrap(buffer);
 	}
 
@@ -109,18 +97,6 @@ public int length() {
 		return this.position;
 	}
 
-	public void pruneBuffer() {
-		clear();
-		if (this.buffer.length > PRUNE_BUFFER_THRESHOLD) {
-			if (LOG.isDebugEnabled()) {
-				LOG.debug("Releasing serialization buffer of " + this.buffer.length + " bytes.");
-			}
-
-			this.buffer = this.startBuffer;
-			this.wrapper = ByteBuffer.wrap(this.buffer);
-		}
-	}
-
 	@Override
 	public String toString() {
 		return String.format("[pos=%d cap=%d]", this.position, this.buffer.length);

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/RecordSerializer.java

@@ -81,13 +81,6 @@ public boolean isFullBuffer() {
 	 */
 	SerializationResult copyToBufferBuilder(BufferBuilder bufferBuilder);
 
-	/**
-	 * Clears the buffer and checks to decrease the size of intermediate data serialization buffer
-	 * after finishing the whole serialization process including
-	 * {@link #serializeRecord(IOReadableWritable)} and {@link #copyToBufferBuilder(BufferBuilder)}.
-	 */
-	void prune();
-
 	/**
 	 * Supports copying an intermediate data serialization buffer to multiple target buffers
 	 * by resetting its initial position before each copying.

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/SpanningRecordSerializer.java

@@ -107,12 +107,6 @@ public void reset() {
 		dataBuffer.position(0);
 	}
 
-	@Override
-	public void prune() {
-		serializationBuffer.pruneBuffer();
-		dataBuffer = serializationBuffer.wrapAsByteBuffer();
-	}
-
 	@Override
 	public boolean hasSerializedData() {
 		return dataBuffer.hasRemaining();

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/ChannelSelectorRecordWriter.java

@@ -75,15 +75,8 @@ public void broadcastEmit(T record) throws IOException, InterruptedException {
 
 		serializer.serializeRecord(record);
 
-		boolean pruneAfterCopying = false;
 		for (int targetChannel = 0; targetChannel < numberOfChannels; targetChannel++) {
-			if (copyFromSerializerToTargetChannel(targetChannel)) {
-				pruneAfterCopying = true;
-			}
-		}
-
-		if (pruneAfterCopying) {
-			serializer.prune();
+			copyFromSerializerToTargetChannel(targetChannel);
 		}
 	}
 

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/RecordWriter.java

@@ -120,9 +120,7 @@ protected void emit(T record, int targetChannel) throws IOException, Interrupted
 		serializer.serializeRecord(record);
 
 		// Make sure we don't hold onto the large intermediate serialization buffer for too long
-		if (copyFromSerializerToTargetChannel(targetChannel)) {
-			serializer.prune();
-		}
+		copyFromSerializerToTargetChannel(targetChannel);
 	}
 
 	/**