[FLINK-6162] [core] Fix bug in ByteArrayOutputStreamWithPos

Fix off-by-one error in 'setPosition' and expand array when seeking
outside the original bounds.

This closes #3595

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

@@ -118,7 +118,8 @@ public class ByteArrayInputStreamWithPos extends InputStream {
 		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 class ByteArrayOutputStreamWithPos extends OutputStream {
 		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

+/*
+ * 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.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+
+public class ByteArrayInputStreamWithPosTest {
+
+	@Rule
+	public ExpectedException thrown = ExpectedException.none();
+
+	private byte[] data = new byte[] {'0','1','2','3','4','5','6','7','8','9'};
+
+	private ByteArrayInputStreamWithPos stream;
+
+	@Before
+	public void setup() {
+		stream = new ByteArrayInputStreamWithPos(data);
+	}
+
+	/**
+	 *  Test setting position on a {@link ByteArrayInputStreamWithPos}
+	 */
+	@Test
+	public void testSetPosition() throws Exception {
+		Assert.assertEquals(data.length, stream.available());
+		Assert.assertEquals('0', stream.read());
+
+		stream.setPosition(1);
+		Assert.assertEquals(data.length - 1, stream.available());
+		Assert.assertEquals('1', stream.read());
+
+		stream.setPosition(3);
+		Assert.assertEquals(data.length - 3, stream.available());
+		Assert.assertEquals('3', stream.read());
+
+		stream.setPosition(data.length);
+		Assert.assertEquals(0, stream.available());
+		Assert.assertEquals(-1, stream.read());
+	}
+
+	/**
+	 * Test 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.");
+		stream.setPosition(data.length + 1);
+	}
+
+	/**
+	 * Test setting a negative position
+	 */
+	@Test
+	public void testSetNegativePosition() throws Exception {
+		thrown.expect(IllegalArgumentException.class);
+		thrown.expectMessage("Position out of bounds.");
+		stream.setPosition(-1);
+	}
+}

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

+/*
+ * 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.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+
+import java.io.IOException;
+import java.util.Arrays;
+
+public class ByteArrayOutputStreamWithPosTest {
+
+	private static final int BUFFER_SIZE = 32;
+
+	@Rule
+	public ExpectedException thrown = ExpectedException.none();
+
+	private ByteArrayOutputStreamWithPos stream;
+
+	@Before
+	public void setup() {
+		stream = new ByteArrayOutputStreamWithPos(BUFFER_SIZE);
+	}
+
+	/**
+	 * Test setting position which is exactly the same with the buffer size
+	 */
+	@Test
+	public void testSetPositionWhenBufferIsFull() throws Exception {
+		stream.write(new byte[BUFFER_SIZE]);
+
+		// check whether the buffer is filled fully
+		Assert.assertEquals(BUFFER_SIZE, stream.getBuf().length);
+
+		// check current position is the end of the buffer
+		Assert.assertEquals(BUFFER_SIZE, stream.getPosition());
+
+		stream.setPosition(BUFFER_SIZE);
+
+		// confirm current position is at where we expect.
+		Assert.assertEquals(BUFFER_SIZE, stream.getPosition());
+	}
+
+	/**
+	 * Test setting negative position
+	 */
+	@Test
+	public void testSetNegativePosition() throws Exception {
+		stream.write(new byte[BUFFER_SIZE]);
+
+		thrown.expect(IllegalArgumentException.class);
+		thrown.expectMessage("Position out of bounds");
+		stream.setPosition(-1);
+	}
+
+	/**
+	 * Test setting position larger than buffer size
+	 */
+	@Test
+	public void testSetPositionLargerThanBufferSize() throws Exception {
+		// fully fill the buffer
+		stream.write(new byte[BUFFER_SIZE]);
+		Assert.assertEquals(BUFFER_SIZE, stream.getBuf().length);
+
+		// expand the buffer by setting position beyond the buffer length
+		stream.setPosition(BUFFER_SIZE + 1);
+		Assert.assertEquals(BUFFER_SIZE * 2, stream.getBuf().length);
+		Assert.assertEquals(BUFFER_SIZE + 1, stream.getPosition());
+	}
+
+	/**
+	 * Test that toString returns a substring of the buffer with range(0, position)
+	 */
+	@Test
+	public void testToString() throws IOException {
+		byte[] data = "1234567890".getBytes(ConfigConstants.DEFAULT_CHARSET);
+
+		ByteArrayOutputStreamWithPos stream = new ByteArrayOutputStreamWithPos(data.length);
+
+		stream.write(data);
+		Assert.assertArrayEquals(data, stream.toString().getBytes(ConfigConstants.DEFAULT_CHARSET));
+
+		for (int i = 0 ; i < data.length ; i++) {
+			stream.setPosition(i);
+			Assert.assertArrayEquals(Arrays.copyOf(data, i), stream.toString().getBytes(ConfigConstants.DEFAULT_CHARSET));
+		}
+
+		// validate that the stored bytes are still tracked properly even when expanding array
+		stream.setPosition(data.length + 1);
+		Assert.assertArrayEquals(Arrays.copyOf(data, data.length + 1), stream.toString().getBytes(ConfigConstants.DEFAULT_CHARSET));
+	}
+}