[FLINK-6162]Fix bug in ByteArrayOutputStreamWithPos#setPosition

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

@@ -118,7 +118,8 @@ public int getPosition() {
 		return position;
 	}
 
-	public void setPos(int pos) {
+	public void setPosition(int pos) {
+		Preconditions.checkArgument(pos >=0 && pos <= count, "Position out of bounds.");
 		this.position = pos;
 	}
 }

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

@@ -101,16 +101,13 @@ public String toString() {
 		return new String(buffer, 0, count, ConfigConstants.DEFAULT_CHARSET);
 	}
 
-	private int getEndPosition() {
-		return buffer.length;
-	}
-
 	public int getPosition() {
 		return count;
 	}
 
 	public void setPosition(int position) {
-		Preconditions.checkArgument(position < getEndPosition(), "Position out of bounds.");
+		Preconditions.checkArgument(position >=0, "Position out of bounds.");
+		ensureCapacity(position + 1);
 		count = position;
 	}
 

flink-core/src/test/java/org/apache/flink/core/memory/ByteArrayInputStreamWithPosTest.java

@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.core.memory;
+
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+
+public class ByteArrayInputStreamWithPosTest {
+
+	@Rule
+	public ExpectedException thrown = ExpectedException.none();
+
+	/**
+	 *  This tests setting position on a {@link ByteArrayInputStreamWithPos}
+	 */
+	@Test
+	public void testSetPosition() throws Exception {
+		byte[] data = new byte[] {'0','1','2','3','4','5','6','7','8','9'};
+		ByteArrayInputStreamWithPos inputStreamWithPos = new ByteArrayInputStreamWithPos(data);
+		inputStreamWithPos.setPosition(1);
+		Assert.assertEquals(data.length - 1, inputStreamWithPos.available());
+		Assert.assertEquals('1', inputStreamWithPos.read());
+		inputStreamWithPos.setPosition(3);
+		Assert.assertEquals(data.length - 3, inputStreamWithPos.available());
+		Assert.assertEquals('3', inputStreamWithPos.read());
+		inputStreamWithPos.setPosition(data.length);
+		Assert.assertEquals(0, inputStreamWithPos.available());
+		Assert.assertEquals(-1, inputStreamWithPos.read());
+	}
+
+	/**
+	 * This tests that the expected position exceeds the capacity of the byte array.
+	 */
+	@Test
+	public void testSetTooLargePosition() throws Exception {
+		thrown.expect(IllegalArgumentException.class);
+		thrown.expectMessage("Position out of bounds.");
+		byte[] data = new byte[] {'0','1','2','3','4','5','6','7','8','9'};
+		ByteArrayInputStreamWithPos inputStreamWithPos = new ByteArrayInputStreamWithPos(data);
+		inputStreamWithPos.setPosition(data.length + 1);
+		Assert.fail("Should not reach here !!!!");
+	}
+
+	/**
+	 * This tests setting a negative position
+	 */
+	@Test
+	public void testSetNegativePosition() throws Exception {
+		thrown.expect(IllegalArgumentException.class);
+		thrown.expectMessage("Position out of bounds.");
+		byte[] data = new byte[] {'0','1','2','3','4','5','6','7','8','9'};
+		ByteArrayInputStreamWithPos inputStreamWithPos = new ByteArrayInputStreamWithPos(data);
+		inputStreamWithPos.setPosition(-1);
+		Assert.fail("Should not reach here !!!!");
+	}
+
+}

flink-core/src/test/java/org/apache/flink/core/memory/ByteArrayOutputStreamWithPosTest.java

@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.core.memory;
+
+import org.apache.flink.configuration.ConfigConstants;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+
+import java.io.IOException;
+import java.util.Arrays;
+
+public class ByteArrayOutputStreamWithPosTest {
+
+	@Rule
+	public ExpectedException thrown = ExpectedException.none();
+
+	/**
+	 * This tests setting position which is exactly the same with the buffer size.
+	 */
+	@Test
+	public void testSetPositionWhenBufferIsFull() throws Exception {
+
+		int initBufferSize = 32;
+
+		ByteArrayOutputStreamWithPos stream = new ByteArrayOutputStreamWithPos(initBufferSize);
+
+		stream.write(new byte[initBufferSize]);
+
+		// check whether the buffer is filled fully
+		Assert.assertEquals(initBufferSize, stream.getBuf().length);
+
+		// check current position is the end of the buffer
+		Assert.assertEquals(initBufferSize, stream.getPosition());
+
+		stream.setPosition(initBufferSize);
+
+		// confirm current position is at where we expect.
+		Assert.assertEquals(initBufferSize, stream.getPosition());
+
+	}
+
+	/**
+	 * This tests setting negative position
+	 */
+	@Test
+	public void testSetNegativePosition() throws Exception {
+		thrown.expect(IllegalArgumentException.class);
+		thrown.expectMessage("Position out of bounds");
+
+		int initBufferSize = 32;
+
+		ByteArrayOutputStreamWithPos stream = new ByteArrayOutputStreamWithPos(initBufferSize);
+
+		stream.write(new byte[initBufferSize]);
+
+		stream.setPosition(-1);
+
+		Assert.fail("Should not reach here !!!!");
+	}
+
+	/**
+	 * This tests setting position larger than buffer size
+	 */
+	@Test
+	public void testSetPositionLargerThanBufferSize() throws Exception {
+		int initBufferSize = 32;
+
+		ByteArrayOutputStreamWithPos stream = new ByteArrayOutputStreamWithPos(initBufferSize);
+
+		stream.write(new byte[initBufferSize]);
+
+		Assert.assertEquals(initBufferSize, stream.getBuf().length);
+
+		stream.setPosition(initBufferSize + 1);
+
+		Assert.assertEquals(initBufferSize * 2, stream.getBuf().length);
+
+		Assert.assertEquals(initBufferSize + 1, stream.getPosition());
+	}
+
+	/**
+	 * THis tests that toString returns a substring of the buffer with range(0, position)
+	 */
+	@Test
+	public void testToString() throws IOException {
+		ByteArrayOutputStreamWithPos stream = new ByteArrayOutputStreamWithPos();
+
+		byte[] data = "1234567890".getBytes(ConfigConstants.DEFAULT_CHARSET);
+		stream.write(data);
+		Assert.assertArrayEquals(data, stream.toString().getBytes(ConfigConstants.DEFAULT_CHARSET));
+
+		stream.setPosition(1);
+		Assert.assertArrayEquals(Arrays.copyOf(data, 1), stream.toString().getBytes(ConfigConstants.DEFAULT_CHARSET));
+
+		stream.setPosition(data.length + 1);
+		Assert.assertArrayEquals(Arrays.copyOf(data, data.length + 1), stream.toString().getBytes(ConfigConstants.DEFAULT_CHARSET));
+
+	}
+}