[hotfix][network] Drop reductant getSizeUnsafe from Buffer interface

The implementations of getSize and getSizeUnsafe are exactly the same now, which do not need the synchronized way.
So We could remove the getSizeUnsafe to make it clean and clear.

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/Buffer.java

@@ -158,17 +158,6 @@ public interface Buffer {
 	 */
 	void setReaderIndex(int readerIndex) throws IndexOutOfBoundsException;
 
-	/**
-	 * Returns the size of the written data, i.e. the <tt>writer index</tt>, of this buffer in an
-	 * non-synchronized fashion.
-	 *
-	 * <p>This is where writable bytes start in the backing memory segment.
-	 *
-	 * @return writer index (from 0 (inclusive) to the size of the backing {@link MemorySegment}
-	 * (inclusive))
-	 */
-	int getSizeUnsafe();
-
 	/**
 	 * Returns the size of the written data, i.e. the <tt>writer index</tt>, of this buffer.
 	 *

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/NetworkBuffer.java

@@ -290,11 +290,6 @@ public void setReaderIndex(int readerIndex) throws IndexOutOfBoundsException {
 		readerIndex(readerIndex);
 	}
 
-	@Override
-	public int getSizeUnsafe() {
-		return writerIndex();
-	}
-
 	@Override
 	public int getSize() {
 		return writerIndex();

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/ReadOnlySlicedNetworkBuffer.java

@@ -152,11 +152,6 @@ public void setReaderIndex(int readerIndex) throws IndexOutOfBoundsException {
 		readerIndex(readerIndex);
 	}
 
-	@Override
-	public int getSizeUnsafe() {
-		return writerIndex();
-	}
-
 	@Override
 	public int getSize() {
 		return writerIndex();

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java

@@ -193,7 +193,7 @@ Optional<BufferAndAvailability> getNextBuffer() throws IOException, InterruptedE
 			}
 		}
 
-		numBytesIn.inc(next.buffer().getSizeUnsafe());
+		numBytesIn.inc(next.buffer().getSize());
 		numBuffersIn.inc();
 		return Optional.of(new BufferAndAvailability(next.buffer(), next.isMoreAvailable(), next.buffersInBacklog()));
 	}

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java

@@ -201,7 +201,7 @@ Optional<BufferAndAvailability> getNextBuffer() throws IOException {
 			moreAvailable = !receivedBuffers.isEmpty();
 		}
 
-		numBytesIn.inc(next.getSizeUnsafe());
+		numBytesIn.inc(next.getSize());
 		numBuffersIn.inc();
 		return Optional.of(new BufferAndAvailability(next, moreAvailable, getSenderBacklog()));
 	}

flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferTest.java

@@ -216,15 +216,13 @@ private static void testCreateSlice1(boolean isBuffer) {
 
 		assertEquals(0, slice.getReaderIndex());
 		assertEquals(10, slice.getSize());
-		assertEquals(10, slice.getSizeUnsafe());
 		assertSame(buffer, slice.unwrap().unwrap());
 
 		// slice indices should be independent:
 		buffer.setSize(8);
 		buffer.setReaderIndex(2);
 		assertEquals(0, slice.getReaderIndex());
 		assertEquals(10, slice.getSize());
-		assertEquals(10, slice.getSizeUnsafe());
 	}
 
 	@Test
@@ -244,15 +242,13 @@ private static void testCreateSlice2(boolean isBuffer) {
 
 		assertEquals(0, slice.getReaderIndex());
 		assertEquals(10, slice.getSize());
-		assertEquals(10, slice.getSizeUnsafe());
 		assertSame(buffer, slice.unwrap().unwrap());
 
 		// slice indices should be independent:
 		buffer.setSize(8);
 		buffer.setReaderIndex(2);
 		assertEquals(0, slice.getReaderIndex());
 		assertEquals(10, slice.getSize());
-		assertEquals(10, slice.getSizeUnsafe());
 	}
 
 	@Test
@@ -312,13 +308,11 @@ private static void testSetGetSize(boolean isBuffer) {
 		NetworkBuffer buffer = newBuffer(1024, 1024, isBuffer);
 
 		assertEquals(0, buffer.getSize()); // initially 0
-		assertEquals(0, buffer.getSizeUnsafe());
 		assertEquals(buffer.writerIndex(), buffer.getSize());
 		assertEquals(0, buffer.readerIndex()); // initially 0
 
 		buffer.setSize(10);
 		assertEquals(10, buffer.getSize());
-		assertEquals(10, buffer.getSizeUnsafe());
 		assertEquals(buffer.writerIndex(), buffer.getSize());
 		assertEquals(0, buffer.readerIndex()); // independent
 	}

flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/ReadOnlySlicedBufferTest.java

@@ -233,8 +233,7 @@ private void testGetSetReaderIndex(ReadOnlySlicedNetworkBuffer slice) {
 	/**
 	 * Tests the independence of the writer index via
 	 * {@link ReadOnlySlicedNetworkBuffer#setSize(int)},
-	 * {@link ReadOnlySlicedNetworkBuffer#getSize()}, and
-	 * {@link ReadOnlySlicedNetworkBuffer#getSizeUnsafe()}.
+	 * {@link ReadOnlySlicedNetworkBuffer#getSize()}.
 	 */
 	@Test
 	public void testGetSetSize1() {
@@ -244,8 +243,7 @@ public void testGetSetSize1() {
 	/**
 	 * Tests the independence of the writer index via
 	 * {@link ReadOnlySlicedNetworkBuffer#setSize(int)},
-	 * {@link ReadOnlySlicedNetworkBuffer#getSize()}, and
-	 * {@link ReadOnlySlicedNetworkBuffer#getSizeUnsafe()}.
+	 * {@link ReadOnlySlicedNetworkBuffer#getSize()}.
 	 */
 	@Test
 	public void testGetSetSize2() {
@@ -254,14 +252,10 @@ public void testGetSetSize2() {
 
 	private void testGetSetSize(ReadOnlySlicedNetworkBuffer slice, int sliceSize) {
 		assertEquals(DATA_SIZE, buffer.getSize());
-		assertEquals(DATA_SIZE, buffer.getSizeUnsafe());
 		assertEquals(sliceSize, slice.getSize());
-		assertEquals(sliceSize, slice.getSizeUnsafe());
 		buffer.setSize(DATA_SIZE + 1);
 		assertEquals(DATA_SIZE + 1, buffer.getSize());
-		assertEquals(DATA_SIZE + 1, buffer.getSizeUnsafe());
 		assertEquals(sliceSize, slice.getSize());
-		assertEquals(sliceSize, slice.getSizeUnsafe());
 	}
 
 	@Test