[FLINK-1320] [core] Add an off-heap variant of the managed memory

This closes #1093

flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java

@@ -125,9 +125,14 @@ public final class ConfigConstants {
 	public static final String TASK_MANAGER_MEMORY_FRACTION_KEY = "taskmanager.memory.fraction";
 
 	/**
-	 * The key for the config parameter defining whether the memory manager allocates memory lazy.
+	 * The fraction of off-heap memory relative to the heap size.
 	 */
-	public static final String TASK_MANAGER_MEMORY_LAZY_ALLOCATION_KEY = "taskmanager.memory.lazyalloc";
+	public static final String TASK_MANAGER_MEMORY_OFF_HEAP_RATIO_KEY = "taskmanager.memory.off-heap-ratio";
+	
+	/**
+	 * The config parameter defining the memory allocation method (JVM heap or off-heap).
+	*/
+	public static final String TASK_MANAGER_MEMORY_OFF_HEAP_KEY = "taskmanager.memory.off-heap";
 
 	/**
 	 * The config parameter defining the number of buffers used in the network stack. This defines the
@@ -542,6 +547,11 @@ public final class ConfigConstants {
 	 */
 	public static final float DEFAULT_MEMORY_MANAGER_MEMORY_FRACTION = 0.7f;
 
+	/**
+	 * The default ratio of heap to off-heap memory, when the TaskManager is started with off-heap memory.
+	 */
+	public static final float DEFAULT_MEMORY_MANAGER_MEMORY_OFF_HEAP_RATIO = 3.0f;
+	
 	/**
 	 * Default number of buffers used in the network stack.
 	 */

flink-core/src/main/java/org/apache/flink/core/memory/HeapMemorySegment.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 java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Objects;
+
+/**
+ * This class represents a piece of heap memory managed by Flink.
+ * The segment is backed by a byte array and features random put and get methods for the basic types,
+ * as well as compare and swap methods.
+ * <p>
+ * This class specialized byte access and byte copy calls for heap memory, while reusing the
+ * multi-byte type accesses and cross-segment operations from the MemorySegment.
+ * <p>
+ * Note that memory segments should usually not be allocated manually, but rather through the
+ * {@link MemorySegmentFactory}.
+ */
+public final class HeapMemorySegment extends MemorySegment {
+
+	/** An extra reference to the heap memory, so we can let byte array checks fail 
+	 *  by the built-in checks automatically without extra checks */
+	private byte[] memory;
+
+	/**
+	 * Creates a new memory segment that represents the data in the given byte array.
+	 * The owner of this memory segment is null.
+	 *
+	 * @param memory The byte array that holds the data.
+	 */
+	HeapMemorySegment(byte[] memory) {
+		this(memory, null);
+	}
+	
+	/**
+	 * Creates a new memory segment that represents the data in the given byte array.
+	 * The memory segment references the given owner.
+	 *
+	 * @param memory The byte array that holds the data.
+	 * @param owner The owner referenced by the memory segment.
+	 */
+	HeapMemorySegment(byte[] memory, Object owner) {
+		super(Objects.requireNonNull(memory), owner);
+		this.memory = memory;
+	}
+	
+	// -------------------------------------------------------------------------
+	//  MemorySegment operations
+	// -------------------------------------------------------------------------
+
+	@Override
+	public void free() {
+		super.free();
+		this.memory = null;
+	}
+
+	@Override
+	public ByteBuffer wrap(int offset, int length) {
+		try {
+			return ByteBuffer.wrap(this.memory, offset, length);
+		}
+		catch (NullPointerException e) {
+			throw new IllegalStateException("segment has been freed");
+		}
+	}
+
+	/**
+	 * Gets the byte array that backs this memory segment.
+	 *
+	 * @return The byte array that backs this memory segment, or null, if the segment has been freed.
+	 */
+	public byte[] getArray() {
+		return this.heapMemory;
+	}
+	
+	// ------------------------------------------------------------------------
+	//                    Random Access get() and put() methods
+	// ------------------------------------------------------------------------
+	
+	@Override
+	public final byte get(int index) {
+		return this.memory[index];
+	}
+
+	@Override
+	public final void put(int index, byte b) {
+		this.memory[index] = b;
+	}
+
+	@Override
+	public final void get(int index, byte[] dst) {
+		get(index, dst, 0, dst.length);
+	}
+
+	@Override
+	public final void put(int index, byte[] src) {
+		put(index, src, 0, src.length);
+	}
+
+	@Override
+	public final void get(int index, byte[] dst, int offset, int length) {
+		// system arraycopy does the boundary checks anyways, no need to check extra
+		System.arraycopy(this.memory, index, dst, offset, length);
+	}
+
+	@Override
+	public final void put(int index, byte[] src, int offset, int length) {
+		// system arraycopy does the boundary checks anyways, no need to check extra
+		System.arraycopy(src, offset, this.memory, index, length);
+	}
+
+	@Override
+	public final boolean getBoolean(int index) {
+		return this.memory[index] != 0;
+	}
+
+	@Override
+	public final void putBoolean(int index, boolean value) {
+		this.memory[index] = (byte) (value ? 1 : 0);
+	}
+
+	// -------------------------------------------------------------------------
+	//                     Bulk Read and Write Methods
+	// -------------------------------------------------------------------------
+
+	@Override
+	public final void get(DataOutput out, int offset, int length) throws IOException {
+		out.write(this.memory, offset, length);
+	}
+
+	@Override
+	public final void put(DataInput in, int offset, int length) throws IOException {
+		in.readFully(this.memory, offset, length);
+	}
+
+	@Override
+	public final void get(int offset, ByteBuffer target, int numBytes) {
+		// ByteBuffer performs the boundary checks
+		target.put(this.memory, offset, numBytes);
+	}
+
+	@Override
+	public final void put(int offset, ByteBuffer source, int numBytes) {
+		// ByteBuffer performs the boundary checks
+		source.get(this.memory, offset, numBytes);
+	}
+
+	// -------------------------------------------------------------------------
+	//                             Factoring
+	// -------------------------------------------------------------------------
+
+	/**
+	 * A memory segment factory that produces heap memory segments. Note that this factory does not
+	 * support to allocate off-heap memory.
+	 */
+	public static final class HeapMemorySegmentFactory implements MemorySegmentFactory.Factory {
+
+		@Override
+		public HeapMemorySegment wrap(byte[] memory) {
+			return new HeapMemorySegment(memory);
+		}
+
+		@Override
+		public HeapMemorySegment allocateUnpooledSegment(int size, Object owner) {
+			return new HeapMemorySegment(new byte[size], owner);
+		}
+
+		@Override
+		public HeapMemorySegment wrapPooledHeapMemory(byte[] memory, Object owner) {
+			return new HeapMemorySegment(memory, owner);
+		}
+
+		@Override
+		public HeapMemorySegment wrapPooledOffHeapMemory(ByteBuffer memory, Object owner) {
+			throw new UnsupportedOperationException(
+					"The MemorySegment factory was not initialized for off-heap memory.");
+		}
+
+		/** prevent external instantiation */
+		HeapMemorySegmentFactory() {}
+	};
+
+	public static final HeapMemorySegmentFactory FACTORY = new HeapMemorySegmentFactory();
+}

flink-core/src/main/java/org/apache/flink/core/memory/HybridMemorySegment.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 java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.nio.BufferOverflowException;
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+
+/**
+ * This class represents a piece of memory managed by Flink. The memory can be on-heap or off-heap,
+ * this is transparently handled by this class.
+ * <p>
+ * This class specialized byte access and byte copy calls for heap memory, while reusing the
+ * multi-byte type accesses and cross-segment operations from the MemorySegment.
+ * <p>
+ * This class subsumes the functionality of the {@link org.apache.flink.core.memory.HeapMemorySegment}, 
+ * but is a bit less efficient for operations on individual bytes.
+ * <p>
+ * Note that memory segments should usually not be allocated manually, but rather through the
+ * {@link MemorySegmentFactory}.
+ */
+public final class HybridMemorySegment extends MemorySegment {
+	
+	/** The direct byte buffer that allocated the off-heap memory. This memory segment holds a reference
+	 * to that buffer, so as long as this memory segment lives, the memory will not be released. */
+	private final ByteBuffer offHeapBuffer;
+
+	/**
+	 * Creates a new memory segment that represents the memory backing the given direct byte buffer.
+	 * Note that the given ByteBuffer must be direct {@link java.nio.ByteBuffer#allocateDirect(int)},
+	 * otherwise this method with throw an IllegalArgumentException.
+	 * <p>
+	 * The owner referenced by this memory segment is null.
+	 *
+	 * @param buffer The byte buffer whose memory is represented by this memory segment.
+	 * @throws IllegalArgumentException Thrown, if the given ByteBuffer is not direct.
+	 */
+	HybridMemorySegment(ByteBuffer buffer) {
+		this(buffer, null);
+	}
+	
+	/**
+	 * Creates a new memory segment that represents the memory backing the given direct byte buffer.
+	 * Note that the given ByteBuffer must be direct {@link java.nio.ByteBuffer#allocateDirect(int)},
+	 * otherwise this method with throw an IllegalArgumentException.
+	 * <p>
+	 * The memory segment references the given owner.
+	 *
+	 * @param buffer The byte buffer whose memory is represented by this memory segment.
+	 * @param owner The owner references by this memory segment.
+	 * @throws IllegalArgumentException Thrown, if the given ByteBuffer is not direct.
+	 */
+	HybridMemorySegment(ByteBuffer buffer, Object owner) {
+		super(checkBufferAndGetAddress(buffer), buffer.capacity(), owner);
+		this.offHeapBuffer = buffer;
+	}
+
+	/**
+	 * Creates a new memory segment that represents the memory of the byte array.
+	 * <p>
+	 * The owner referenced by this memory segment is null.
+	 *
+	 * @param buffer The byte array whose memory is represented by this memory segment.
+	 */
+	HybridMemorySegment(byte[] buffer) {
+		this(buffer, null);
+	}
+
+	/**
+	 * Creates a new memory segment that represents the memory of the byte array.
+	 * <p>
+	 * The memory segment references the given owner.
+	 *
+	 * @param buffer The byte array whose memory is represented by this memory segment.
+	 * @param owner The owner references by this memory segment.
+	 */
+	HybridMemorySegment(byte[] buffer, Object owner) {
+		super(buffer, owner);
+		this.offHeapBuffer = null;
+	}
+
+	// -------------------------------------------------------------------------
+	//  MemorySegment operations
+	// -------------------------------------------------------------------------
+	
+	public byte[] getArray() {
+		if (heapMemory != null) {
+			return heapMemory;
+		} else {
+			throw new IllegalStateException("Memory segment does not represent heap memory");
+		}
+	}
+
+	/**
+	 * Gets the buffer that owns the memory of this memory segment.
+	 *
+	 * @return The byte buffer that owns the memory of this memory segment.
+	 */
+	public ByteBuffer getOffHeapBuffer() {
+		if (offHeapBuffer != null) {
+			return offHeapBuffer;
+		} else {
+			throw new IllegalStateException("Memory segment does not represent off heap memory");
+		}
+	}
+
+	@Override
+	public ByteBuffer wrap(int offset, int length) {
+		if (address <= addressLimit) {
+			if (heapMemory != null) {
+				return ByteBuffer.wrap(heapMemory, offset, length);
+			}
+			else {
+				try {
+					ByteBuffer wrapper = offHeapBuffer.duplicate();
+					wrapper.limit(offset + length);
+					wrapper.position(offset);
+					return wrapper;
+				}
+				catch (IllegalArgumentException e) {
+					throw new IndexOutOfBoundsException();
+				}
+			}
+		}
+		else {
+			throw new IllegalStateException("segment has been freed");
+		}
+	}
+
+	// ------------------------------------------------------------------------
+	//  Random Access get() and put() methods
+	// ------------------------------------------------------------------------
+	
+	@Override
+	public byte get(int index) {
+		final long pos = address + index;
+		if (index >= 0 && pos < addressLimit) {
+			return UNSAFE.getByte(heapMemory, pos);
+		}
+		else if (address > addressLimit) {
+			throw new IllegalStateException("segment has been freed");
+		}
+		else {
+			// index is in fact invalid
+			throw new IndexOutOfBoundsException();
+		}
+	}
+
+	@Override
+	public void put(int index, byte b) {
+		final long pos = address + index;
+		if (index >= 0 && pos < addressLimit) {
+			UNSAFE.putByte(heapMemory, pos, b);
+		}
+		else if (address > addressLimit) {
+			throw new IllegalStateException("segment has been freed");
+		}
+		else {
+			// index is in fact invalid
+			throw new IndexOutOfBoundsException();
+		}
+	}
+
+	@Override
+	public void get(int index, byte[] dst) {
+		get(index, dst, 0, dst.length);
+	}
+
+	@Override
+	public void put(int index, byte[] src) {
+		put(index, src, 0, src.length);
+	}
+
+	@Override
+	public void get(int index, byte[] dst, int offset, int length) {
+		// check the byte array offset and length and the status
+		if ( (offset | length | (offset + length) | (dst.length - (offset + length))) < 0) {
+			throw new IndexOutOfBoundsException();
+		}
+
+		final long pos = address + index;
+		if (index >= 0 && pos <= addressLimit - length) {
+			final long arrayAddress = BYTE_ARRAY_BASE_OFFSET + offset;
+			UNSAFE.copyMemory(heapMemory, pos, dst, arrayAddress, length);
+		}
+		else if (address > addressLimit) {
+			throw new IllegalStateException("segment has been freed");
+		}
+		else {
+			// index is in fact invalid
+			throw new IndexOutOfBoundsException();
+		}
+	}
+
+	@Override
+	public void put(int index, byte[] src, int offset, int length) {
+		// check the byte array offset and length
+		if ((offset | length | (offset + length) | (src.length - (offset + length))) < 0) {
+			throw new IndexOutOfBoundsException();
+		}
+		
+		final long pos = address + index;
+
+		if (index >= 0 && pos <= addressLimit - length) {
+			final long arrayAddress = BYTE_ARRAY_BASE_OFFSET + offset;
+			UNSAFE.copyMemory(src, arrayAddress, heapMemory, pos, length);
+		}
+		else if (address > addressLimit) {
+			throw new IllegalStateException("segment has been freed");
+		}
+		else {
+			// index is in fact invalid
+			throw new IndexOutOfBoundsException();
+		}
+	}
+
+	@Override
+	public boolean getBoolean(int index) {
+		return get(index) != 0;
+	}
+
+	@Override
+	public void putBoolean(int index, boolean value) {
+		put(index, (byte) (value ? 1 : 0));
+	}
+
+	// -------------------------------------------------------------------------
+	//  Bulk Read and Write Methods
+	// -------------------------------------------------------------------------
+
+	@Override
+	public final void get(DataOutput out, int offset, int length) throws IOException {
+		if (address <= addressLimit) {
+			if (heapMemory != null) {
+				out.write(heapMemory, offset, length);
+			}
+			else {
+				while (length >= 8) {
+					out.writeLong(getLongBigEndian(offset));
+					offset += 8;
+					length -= 8;
+				}
+		
+				while (length > 0) {
+					out.writeByte(get(offset));
+					offset++;
+					length--;
+				}
+			}
+		}
+		else {
+			throw new IllegalStateException("segment has been freed");
+		}
+	}
+
+	@Override
+	public final void put(DataInput in, int offset, int length) throws IOException {
+		if (address <= addressLimit) {
+			if (heapMemory != null) {
+				in.readFully(heapMemory, offset, length);
+			}
+			else {
+				while (length >= 8) {
+					putLongBigEndian(offset, in.readLong());
+					offset += 8;
+					length -= 8;
+				}
+				while (length > 0) {
+					put(offset, in.readByte());
+					offset++;
+					length--;
+				}
+			}
+		}
+		else {
+			throw new IllegalStateException("segment has been freed");
+		}
+	}
+
+	@Override
+	public final void get(int offset, ByteBuffer target, int numBytes) {
+		// check the byte array offset and length
+		if ((offset | numBytes | (offset + numBytes)) < 0) {
+			throw new IndexOutOfBoundsException();
+		}
+
+		final int targetOffset = target.position();
+		final int remaining = target.remaining();
+
+		if (remaining < numBytes) {
+			throw new BufferOverflowException();
+		}
+
+		if (target.isDirect()) {
+			// copy to the target memory directly
+			final long targetPointer = getAddress(target) + targetOffset;
+			final long sourcePointer = address + offset;
+
+			if (sourcePointer <= addressLimit - numBytes) {
+				UNSAFE.copyMemory(heapMemory, sourcePointer, null, targetPointer, numBytes);
+				target.position(targetOffset + numBytes);
+			}
+			else if (address > addressLimit) {
+				throw new IllegalStateException("segment has been freed");
+			}
+			else {
+				throw new IndexOutOfBoundsException();
+			}
+		}
+		else if (target.hasArray()) {
+			// move directly into the byte array
+			get(offset, target.array(), targetOffset + target.arrayOffset(), numBytes);
+
+			// this must be after the get() call to ensue that the byte buffer is not
+			// modified in case the call fails
+			target.position(targetOffset + numBytes);
+		}
+		else {
+			// neither heap buffer nor direct buffer
+			while (target.hasRemaining()) {
+				target.put(get(offset++));
+			}
+		}
+	}
+
+	@Override
+	public final void put(int offset, ByteBuffer source, int numBytes) {
+		// check the byte array offset and length
+		if ((offset | numBytes | (offset + numBytes)) < 0) {
+			throw new IndexOutOfBoundsException();
+		}
+
+		final int sourceOffset = source.position();
+		final int remaining = source.remaining();
+
+		if (remaining < numBytes) {
+			throw new BufferUnderflowException();
+		}
+
+		if (source.isDirect()) {
+			// copy to the target memory directly
+			final long sourcePointer = getAddress(source) + sourceOffset;
+			final long targetPointer = address + offset;
+
+			if (targetPointer <= addressLimit - numBytes) {
+				UNSAFE.copyMemory(null, sourcePointer, heapMemory, targetPointer, numBytes);
+				source.position(sourceOffset + numBytes);
+			}
+			else if (address > addressLimit) {
+				throw new IllegalStateException("segment has been freed");
+			}
+			else {
+				throw new IndexOutOfBoundsException();
+			}
+		}
+		else if (source.hasArray()) {
+			// move directly into the byte array
+			put(offset, source.array(), sourceOffset + source.arrayOffset(), numBytes);
+
+			// this must be after the get() call to ensue that the byte buffer is not
+			// modified in case the call fails
+			source.position(sourceOffset + numBytes);
+		}
+		else {
+			// neither heap buffer nor direct buffer
+			while (source.hasRemaining()) {
+				put(offset++, source.get());
+			}
+		}
+	}
+
+	// --------------------------------------------------------------------------------------------
+	//  Utilities for native memory accesses and checks
+	// --------------------------------------------------------------------------------------------
+
+	/** The reflection fields with which we access the off-heap pointer from direct ByteBuffers */
+	private static final Field ADDRESS_FIELD;
+
+	static {
+		try {
+			ADDRESS_FIELD = java.nio.Buffer.class.getDeclaredField("address");
+			ADDRESS_FIELD.setAccessible(true);
+		}
+		catch (Throwable t) {
+			throw new RuntimeException(
+					"Cannot initialize HybridMemorySegment: off-heap memory is incompatible with this JVM.", t);
+		}
+	}
+
+	private static long getAddress(ByteBuffer buffer) {
+		if (buffer == null) {
+			throw new NullPointerException("buffer is null");
+		}
+		try {
+			return (Long) ADDRESS_FIELD.get(buffer);
+		}
+		catch (Throwable t) {
+			throw new RuntimeException("Could not access direct byte buffer address.", t);
+		}
+	}
+	
+	private static long checkBufferAndGetAddress(ByteBuffer buffer) {
+		if (buffer == null) {
+			throw new NullPointerException("buffer is null");
+		}
+		if (!buffer.isDirect()) {
+			throw new IllegalArgumentException("Can't initialize from non-direct ByteBuffer.");
+		}
+		return getAddress(buffer);
+	}
+
+	// -------------------------------------------------------------------------
+	//  Factoring
+	// -------------------------------------------------------------------------
+
+	/**
+	 * Base factory for hybrid memory segments.
+	 */
+	public static final class HybridMemorySegmentFactory implements MemorySegmentFactory.Factory {
+		
+		@Override
+		public HybridMemorySegment wrap(byte[] memory) {
+			return new HybridMemorySegment(memory);
+		}
+
+		@Override
+		public HybridMemorySegment allocateUnpooledSegment(int size, Object owner) {
+			return new HybridMemorySegment(new byte[size], owner);
+		}
+
+		@Override
+		public HybridMemorySegment wrapPooledHeapMemory(byte[] memory, Object owner) {
+			return new HybridMemorySegment(memory, owner);
+		}
+
+		@Override
+		public HybridMemorySegment wrapPooledOffHeapMemory(ByteBuffer memory, Object owner) {
+			return new HybridMemorySegment(memory, owner);
+		}
+
+		/** prevent external instantiation */
+		HybridMemorySegmentFactory() {}
+	};
+
+	public static final HybridMemorySegmentFactory FACTORY = new HybridMemorySegmentFactory();
+}

flink-core/src/main/java/org/apache/flink/core/memory/MemorySegmentFactory.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 java.nio.ByteBuffer;
+
+/**
+ * A factory for memory segments. The purpose of this factory is to make sure that all memory segments
+ * for heap data are of the same type. That way, the runtime does not mix the various specializations
+ * of the {@link org.apache.flink.core.memory.MemorySegment}. Not mixing them has shown to be beneficial
+ * to method specialization by the JIT and to overall performance.
+ * <p>
+ * Note that this factory auto-initialized to use {@link org.apache.flink.core.memory.HeapMemorySegment},
+ * if a request to create a segment comes before the initialization.
+ */
+public class MemorySegmentFactory {
+
+	/** The factory to use */
+	private static volatile Factory factory;
+	
+	/**
+	 * Creates a new memory segment that targets the given heap memory region.
+	 * This method should be used to turn short lived byte arrays into memory segments.
+	 *
+	 * @param buffer The heap memory region.
+	 * @return A new memory segment that targets the given heap memory region.
+	 */
+	public static MemorySegment wrap(byte[] buffer) {
+		ensureInitialized();
+		return factory.wrap(buffer);
+	}
+
+	/**
+	 * Allocates some unpooled memory and creates a new memory segment that represents
+	 * that memory.
+	 * <p>
+	 * This method is similar to {@link #allocateUnpooledSegment(int, Object)}, but the
+	 * memory segment will have null as the owner.
+	 *
+	 * @param size The size of the memory segment to allocate.
+	 * @return A new memory segment, backed by unpooled heap memory.
+	 */
+	public static MemorySegment allocateUnpooledSegment(int size) {
+		return allocateUnpooledSegment(size, null);
+	}
+
+	/**
+	 * Allocates some unpooled memory and creates a new memory segment that represents
+	 * that memory.
+	 * <p>
+	 * This method is similar to {@link #allocateUnpooledSegment(int)}, but additionally sets
+	 * the owner of the memory segment.
+	 * 
+	 * @param size The size of the memory segment to allocate.
+	 * @param owner The owner to associate with the memory segment.
+	 * @return A new memory segment, backed by unpooled heap memory.
+	 */
+	public static MemorySegment allocateUnpooledSegment(int size, Object owner) {
+		ensureInitialized();
+		return factory.allocateUnpooledSegment(size, owner);
+	}
+
+	/**
+	 * Creates a memory segment that wraps the given byte array.
+	 * <p>
+	 * This method is intended to be used for components which pool memory and create
+	 * memory segments around long-lived memory regions.
+	 *
+	 * 
+	 * @param memory The heap memory to be represented by the memory segment.
+	 * @param owner The owner to associate with the memory segment.
+	 * @return A new memory segment representing the given heap memory.
+	 */
+	public static MemorySegment wrapPooledHeapMemory(byte[] memory, Object owner) {
+		ensureInitialized();
+		return factory.wrapPooledHeapMemory(memory, owner);
+	}
+
+	/**
+	 * Creates a memory segment that wraps the off-heap memory backing the given ByteBuffer.
+	 * Note that the ByteBuffer needs to be a <i>direct ByteBuffer</i>. 
+	 * <p>
+	 * This method is intended to be used for components which pool memory and create
+	 * memory segments around long-lived memory regions.
+	 *
+	 * @param memory The byte buffer with the off-heap memory to be represented by the memory segment.
+	 * @param owner The owner to associate with the memory segment.
+	 * @return A new memory segment representing the given off-heap memory.
+	 */
+	public static MemorySegment wrapPooledOffHeapMemory(ByteBuffer memory, Object owner) {
+		ensureInitialized();
+		return factory.wrapPooledOffHeapMemory(memory, owner);
+	}
+	
+	// ------------------------------------------------------------------------
+	
+	/**
+	 * Initializes this factory with the given concrete factory.
+	 * 
+	 * @param f The concrete factory to use.
+	 * @throws java.lang.IllegalStateException Thrown, if this factory has been initialized before.
+	 */
+	public static void initializeFactory(Factory f) {
+		if (f == null) {
+			throw new NullPointerException();
+		}
+	
+		synchronized (MemorySegmentFactory.class) {
+			if (factory == null) {
+				factory = f;
+			}
+			else {
+				throw new IllegalStateException("Factory has already been initialized");
+			}
+		}
+	}
+
+	/**
+	 * Checks whether this memory segment factory has been initialized (with a type to produce).
+	 * 
+	 * @return True, if the factory has been initialized, false otherwise.
+	 */
+	public static boolean isInitialized() {
+		return factory != null;
+	}
+
+	/**
+	 * Gets the factory. May return null, if the factory has not been initialized.
+	 * 
+	 * @return The factory, or null, if the factory has not been initialized.
+	 */
+	public static Factory getFactory() {
+		return factory;
+	}
+	
+	private static void ensureInitialized() {
+		if (factory == null) {
+			factory = HeapMemorySegment.FACTORY;
+		}
+	}
+
+	// ------------------------------------------------------------------------
+	//  Internal factory
+	// ------------------------------------------------------------------------
+	
+	/**
+	 * A concrete factory for memory segments.
+	 */
+	public static interface Factory {
+
+		/**
+		 * Creates a new memory segment that targets the given heap memory region.
+		 *
+		 * @param memory The heap memory region.
+		 * @return A new memory segment that targets the given heap memory region.
+		 */
+		MemorySegment wrap(byte[] memory);
+
+		/**
+		 * Allocates some unpooled memory and creates a new memory segment that represents
+		 * that memory.
+		 *
+		 * @param size The size of the memory segment to allocate.
+		 * @param owner The owner to associate with the memory segment.
+		 * @return A new memory segment, backed by unpooled heap memory.
+		 */
+		MemorySegment allocateUnpooledSegment(int size, Object owner);
+
+		/**
+		 * Creates a memory segment that wraps the given byte array.
+		 * <p>
+		 * This method is intended to be used for components which pool memory and create
+		 * memory segments around long-lived memory regions.
+		 *
+		 *
+		 * @param memory The heap memory to be represented by the memory segment.
+		 * @param owner The owner to associate with the memory segment.
+		 * @return A new memory segment representing the given heap memory.
+		 */
+		MemorySegment wrapPooledHeapMemory(byte[] memory, Object owner);
+
+		/**
+		 * Creates a memory segment that wraps the off-heap memory backing the given ByteBuffer.
+		 * Note that the ByteBuffer needs to be a <i>direct ByteBuffer</i>. 
+		 * <p>
+		 * This method is intended to be used for components which pool memory and create
+		 * memory segments around long-lived memory regions.
+		 *
+		 * @param memory The byte buffer with the off-heap memory to be represented by the memory segment.
+		 * @param owner The owner to associate with the memory segment.
+		 * @return A new memory segment representing the given off-heap memory.
+		 */
+		MemorySegment wrapPooledOffHeapMemory(ByteBuffer memory, Object owner);
+	}
+}

flink-core/src/main/java/org/apache/flink/core/memory/MemoryType.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;
+
+/**
+ * The class of memory, such as heap or off-heap.
+ */
+public enum MemoryType {
+
+	/**
+	 * Denotes memory that is part of the Java heap.
+	 */
+	HEAP,
+
+	/**
+	 * Denotes memory that is outside the Java heap (but still part of tha Java process).
+	 */
+	OFF_HEAP
+}

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

@@ -22,22 +22,17 @@ import java.lang.reflect.Field;
 import java.nio.ByteOrder;
 
 /**
- * Utility class for native (unsafe) memory accesses.
+ * Utility class for memory operations.
  */
 public class MemoryUtils {
 	
-	/**
-	 * The "unsafe", which can be used to perform native memory accesses.
-	 */
+	/** The "unsafe", which can be used to perform native memory accesses. */
 	@SuppressWarnings("restriction")
 	public static final sun.misc.Unsafe UNSAFE = getUnsafe();
 	
-	/**
-	 * The native byte order of the platform on which the system currently runs.
-	 */
-	public static final ByteOrder NATIVE_BYTE_ORDER = getByteOrder();
-	
-	
+	/** The native byte order of the platform on which the system currently runs. */
+	public static final ByteOrder NATIVE_BYTE_ORDER = ByteOrder.nativeOrder();
+
 	@SuppressWarnings("restriction")
 	private static sun.misc.Unsafe getUnsafe() {
 		try {
@@ -45,21 +40,18 @@ public class MemoryUtils {
 			unsafeField.setAccessible(true);
 			return (sun.misc.Unsafe) unsafeField.get(null);
 		} catch (SecurityException e) {
-			throw new RuntimeException("Could not access the unsafe handle.", e);
+			throw new RuntimeException("Could not access the sun.misc.Unsafe handle, permission denied by security manager.", e);
 		} catch (NoSuchFieldException e) {
-			throw new RuntimeException("The static unsafe handle field was not be found.");
+			throw new RuntimeException("The static handle field in sun.misc.Unsafe was not found.");
 		} catch (IllegalArgumentException e) {
-			throw new RuntimeException("Bug: Illegal argument reflection access for static field.");
+			throw new RuntimeException("Bug: Illegal argument reflection access for static field.", e);
 		} catch (IllegalAccessException e) {
-			throw new RuntimeException("Access to the unsafe handle is forbidden by the runtime.", e);
+			throw new RuntimeException("Access to sun.misc.Unsafe is forbidden by the runtime.", e);
+		} catch (Throwable t) {
+			throw new RuntimeException("Unclassified error while trying to access the sun.misc.Unsafe handle.", t);
 		}
 	}
-	
-	@SuppressWarnings("restriction")
-	private static ByteOrder getByteOrder() {
-		return ByteOrder.nativeOrder();
-	}
-	
-	
+
+	/** Should not be instantiated */
 	private MemoryUtils() {}
 }

flink-core/src/test/java/org/apache/flink/api/common/typeutils/ComparatorTestBase.java

@@ -29,7 +29,9 @@ import static org.junit.Assert.*;
 import org.apache.flink.core.memory.DataInputView;
 import org.apache.flink.core.memory.DataOutputView;
 import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.core.memory.MemorySegmentFactory;
 import org.apache.flink.util.TestLogger;
+
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -238,7 +240,7 @@ public abstract class ComparatorTestBase<T> extends TestLogger {
 	
 	// Help Function for setting up a memory segment and normalize the keys of the data array in it
 	public MemorySegment setupNormalizedKeysMemSegment(T[] data, int normKeyLen, TypeComparator<T> comparator) {
-		MemorySegment memSeg = new MemorySegment(new byte[2048]);
+		MemorySegment memSeg = MemorySegmentFactory.allocateUnpooledSegment(2048);
 
 		// Setup normalized Keys in the memory segment
 		int offset = 0;
@@ -294,7 +296,7 @@ public abstract class ComparatorTestBase<T> extends TestLogger {
 			MemorySegment memSeg2 = setupNormalizedKeysMemSegment(data, normKeyLen, comparator);
 
 			for (int i = 0; i < data.length; i++) {
-				assertTrue(MemorySegment.compare(memSeg1, memSeg2, i * normKeyLen, i * normKeyLen, normKeyLen) == 0);
+				assertTrue(memSeg1.compare(memSeg2, i * normKeyLen, i * normKeyLen, normKeyLen) == 0);
 			}
 		} catch (Exception e) {
 			System.err.println(e.getMessage());
@@ -343,14 +345,14 @@ public abstract class ComparatorTestBase<T> extends TestLogger {
 				for (int h = l + 1; h < data.length; h++) {
 					int cmp;
 					if (greater) {
-						cmp = MemorySegment.compare(memSegLow, memSegHigh, l * normKeyLen, h * normKeyLen, normKeyLen);
+						cmp = memSegLow.compare(memSegHigh, l * normKeyLen, h * normKeyLen, normKeyLen);
 						if (fullyDetermines) {
 							assertTrue(cmp < 0);
 						} else {
 							assertTrue(cmp <= 0);
 						}
 					} else {
-						cmp = MemorySegment.compare(memSegHigh, memSegLow, h * normKeyLen, l * normKeyLen, normKeyLen);
+						cmp = memSegHigh.compare(memSegLow, h * normKeyLen, l * normKeyLen, normKeyLen);
 						if (fullyDetermines) {
 							assertTrue(cmp > 0);
 						} else {

flink-core/src/test/java/org/apache/flink/core/memory/CrossSegmentTypeTest.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.Test;
+
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.Random;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+public class CrossSegmentTypeTest {
+
+	private final int pageSize = 32*1024;
+	
+	// ------------------------------------------------------------------------
+	
+	@Test
+	public void testCompareBytesMixedSegments() {
+		try {
+			MemorySegment[] segs1 = {
+					new HeapMemorySegment(new byte[pageSize]),
+					new HybridMemorySegment(new byte[pageSize]),
+					new HybridMemorySegment(ByteBuffer.allocateDirect(pageSize))
+			};
+
+			MemorySegment[] segs2 = {
+					new HeapMemorySegment(new byte[pageSize]),
+					new HybridMemorySegment(new byte[pageSize]),
+					new HybridMemorySegment(ByteBuffer.allocateDirect(pageSize))
+			};
+		
+			Random rnd = new Random();
+			
+			for (MemorySegment seg1 : segs1) {
+				for (MemorySegment seg2 : segs2) {
+					testCompare(seg1, seg2, rnd);
+				}
+			}
+		}
+		catch (Exception e) {
+			e.printStackTrace();
+			fail(e.getMessage());
+		}
+	}
+	
+	private void testCompare(MemorySegment seg1, MemorySegment seg2, Random random) {
+		assertEquals(pageSize, seg1.size());
+		assertEquals(pageSize, seg2.size());
+		
+		final byte[] bytes1 = new byte[pageSize];
+		final byte[] bytes2 = new byte[pageSize];
+
+		final int stride = pageSize / 255;
+		final int shift = 16666;
+
+		for (int i = 0; i < pageSize; i++) {
+			byte val = (byte) ((i / stride) & 0xff);
+			bytes1[i] = val;
+
+			if (i + shift < bytes2.length) {
+				bytes2[i + shift] = val;
+			}
+		}
+		
+		seg1.put(0, bytes1);
+		seg2.put(0, bytes2);
+
+		for (int i = 0; i < 1000; i++) {
+			int pos1 = random.nextInt(bytes1.length);
+			int pos2 = random.nextInt(bytes2.length);
+
+			int len = Math.min(Math.min(bytes1.length - pos1, bytes2.length - pos2),
+					random.nextInt(pageSize / 50 ));
+
+			int cmp = seg1.compare(seg2, pos1, pos2, len);
+
+			if (pos1 < pos2 - shift) {
+				assertTrue(cmp <= 0);
+			}
+			else {
+				assertTrue(cmp >= 0);
+			}
+		}
+	}
+
+
+	@Test
+	public void testSwapBytesMixedSegments() {
+		try {
+			final int HALF_SIZE = pageSize / 2;
+			
+			MemorySegment[] segs1 = {
+					new HeapMemorySegment(new byte[pageSize]),
+					new HybridMemorySegment(new byte[pageSize]),
+					new HybridMemorySegment(ByteBuffer.allocateDirect(pageSize))
+			};
+
+			MemorySegment[] segs2 = {
+					new HeapMemorySegment(new byte[HALF_SIZE]),
+					new HybridMemorySegment(new byte[HALF_SIZE]),
+					new HybridMemorySegment(ByteBuffer.allocateDirect(HALF_SIZE))
+			};
+
+			Random rnd = new Random();
+
+			for (MemorySegment seg1 : segs1) {
+				for (MemorySegment seg2 : segs2) {
+					testSwap(seg1, seg2, rnd, HALF_SIZE);
+				}
+			}
+		}
+		catch (Exception e) {
+			e.printStackTrace();
+			fail(e.getMessage());
+		}
+	}
+	
+	private void testSwap(MemorySegment seg1, MemorySegment seg2, Random random, int smallerSize) {
+		assertEquals(pageSize, seg1.size());
+		assertEquals(smallerSize, seg2.size());
+
+		final byte[] bytes1 = new byte[pageSize];
+		final byte[] bytes2 = new byte[smallerSize];
+
+		Arrays.fill(bytes2, (byte) 1);
+		
+		seg1.put(0, bytes1);
+		seg2.put(0, bytes2);
+
+		// wap the second half of the first segment with the second segment
+
+		int pos = 0;
+		while (pos < smallerSize) {
+			int len = random.nextInt(pageSize / 40);
+			len = Math.min(len, smallerSize - pos);
+			seg1.swapBytes(new byte[len], seg2, pos + smallerSize, pos, len);
+			pos += len;
+		}
+
+		// the second segment should now be all zeros, the first segment should have one in its second half
+
+		for (int i = 0; i < smallerSize; i++) {
+			assertEquals((byte) 0, seg1.get(i));
+			assertEquals((byte) 0, seg2.get(i));
+			assertEquals((byte) 1, seg1.get(i + smallerSize));
+		}
+	}
+
+	@Test
+	public void testCopyMixedSegments() {
+		try {
+			MemorySegment[] segs1 = {
+					new HeapMemorySegment(new byte[pageSize]),
+					new HybridMemorySegment(new byte[pageSize]),
+					new HybridMemorySegment(ByteBuffer.allocateDirect(pageSize))
+			};
+
+			MemorySegment[] segs2 = {
+					new HeapMemorySegment(new byte[pageSize]),
+					new HybridMemorySegment(new byte[pageSize]),
+					new HybridMemorySegment(ByteBuffer.allocateDirect(pageSize))
+			};
+
+			Random rnd = new Random();
+
+			for (MemorySegment seg1 : segs1) {
+				for (MemorySegment seg2 : segs2) {
+					testCopy(seg1, seg2, rnd);
+				}
+			}
+		}
+		catch (Exception e) {
+			e.printStackTrace();
+			fail(e.getMessage());
+		}
+	}
+
+	private void testCopy(MemorySegment seg1, MemorySegment seg2, Random random) {
+		assertEquals(pageSize, seg1.size());
+		assertEquals(pageSize, seg2.size());
+
+		byte[] expected = new byte[pageSize];
+		byte[] actual = new byte[pageSize];
+		
+		// zero out the memory
+		seg1.put(0, expected);
+		seg2.put(0, expected);
+		
+		for (int i = 0; i < 40; i++) {
+			int numBytes = random.nextInt(pageSize / 20);
+			byte[] bytes = new byte[numBytes];
+			random.nextBytes(bytes);
+			
+			int thisPos = random.nextInt(pageSize - numBytes);
+			int otherPos = random.nextInt(pageSize - numBytes);
+			
+			// track what we expect
+			System.arraycopy(bytes, 0, expected, otherPos, numBytes);
+			
+			seg1.put(thisPos, bytes);
+			seg1.copyTo(thisPos, seg2, otherPos, numBytes);
+		}
+		
+		seg2.get(0, actual);
+		assertArrayEquals(expected, actual);
+		
+		// test out of bound conditions
+
+		final int[] validOffsets = { 0, 1, pageSize / 10 * 9 };
+		final int[] invalidOffsets = { -1, pageSize + 1, -pageSize, Integer.MAX_VALUE, Integer.MIN_VALUE };
+
+		final int[] validLengths = { 0, 1, pageSize / 10, pageSize };
+		final int[] invalidLengths = { -1, -pageSize, pageSize + 1, Integer.MAX_VALUE, Integer.MIN_VALUE };
+
+		for (int off1 : validOffsets) {
+			for (int off2 : validOffsets) {
+				for (int len : invalidLengths) {
+					try {
+						seg1.copyTo(off1, seg2, off2, len);
+						fail("should fail with an IndexOutOfBoundsException");
+					}
+					catch (IndexOutOfBoundsException ignored) {}
+
+					try {
+						seg1.copyTo(off2, seg2, off1, len);
+						fail("should fail with an IndexOutOfBoundsException");
+					}
+					catch (IndexOutOfBoundsException ignored) {}
+					
+					try {
+						seg2.copyTo(off1, seg1, off2, len);
+						fail("should fail with an IndexOutOfBoundsException");
+					}
+					catch (IndexOutOfBoundsException ignored) {}
+
+					try {
+						seg2.copyTo(off2, seg1, off1, len);
+						fail("should fail with an IndexOutOfBoundsException");
+					}
+					catch (IndexOutOfBoundsException ignored) {}
+				}
+			}
+		}
+
+		for (int off1 : validOffsets) {
+			for (int off2 : invalidOffsets) {
+				for (int len : validLengths) {
+					try {
+						seg1.copyTo(off1, seg2, off2, len);
+						fail("should fail with an IndexOutOfBoundsException");
+					}
+					catch (IndexOutOfBoundsException ignored) {}
+
+					try {
+						seg1.copyTo(off2, seg2, off1, len);
+						fail("should fail with an IndexOutOfBoundsException");
+					}
+					catch (IndexOutOfBoundsException ignored) {}
+
+					try {
+						seg2.copyTo(off1, seg1, off2, len);
+						fail("should fail with an IndexOutOfBoundsException");
+					}
+					catch (IndexOutOfBoundsException ignored) {}
+
+					try {
+						seg2.copyTo(off2, seg1, off1, len);
+						fail("should fail with an IndexOutOfBoundsException");
+					}
+					catch (IndexOutOfBoundsException ignored) {}
+				}
+			}
+		}
+
+		for (int off1 : invalidOffsets) {
+			for (int off2 : validOffsets) {
+				for (int len : validLengths) {
+					try {
+						seg1.copyTo(off1, seg2, off2, len);
+						fail("should fail with an IndexOutOfBoundsException");
+					}
+					catch (IndexOutOfBoundsException ignored) {}
+
+					try {
+						seg1.copyTo(off2, seg2, off1, len);
+						fail("should fail with an IndexOutOfBoundsException");
+					}
+					catch (IndexOutOfBoundsException ignored) {}
+
+					try {
+						seg2.copyTo(off1, seg1, off2, len);
+						fail("should fail with an IndexOutOfBoundsException");
+					}
+					catch (IndexOutOfBoundsException ignored) {}
+
+					try {
+						seg2.copyTo(off2, seg1, off1, len);
+						fail("should fail with an IndexOutOfBoundsException");
+					}
+					catch (IndexOutOfBoundsException ignored) {}
+				}
+			}
+		}
+
+		for (int off1 : invalidOffsets) {
+			for (int off2 : invalidOffsets) {
+				for (int len : validLengths) {
+					try {
+						seg1.copyTo(off1, seg2, off2, len);
+						fail("should fail with an IndexOutOfBoundsException");
+					}
+					catch (IndexOutOfBoundsException ignored) {}
+
+					try {
+						seg1.copyTo(off2, seg2, off1, len);
+						fail("should fail with an IndexOutOfBoundsException");
+					}
+					catch (IndexOutOfBoundsException ignored) {}
+
+					try {
+						seg2.copyTo(off1, seg1, off2, len);
+						fail("should fail with an IndexOutOfBoundsException");
+					}
+					catch (IndexOutOfBoundsException ignored) {}
+
+					try {
+						seg2.copyTo(off2, seg1, off1, len);
+						fail("should fail with an IndexOutOfBoundsException");
+					}
+					catch (IndexOutOfBoundsException ignored) {}
+				}
+			}
+		}
+	}
+}

flink-core/src/test/java/org/apache/flink/core/memory/EndiannessAccessChecks.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 java.nio.ByteBuffer;
+import java.util.Random;
+import org.junit.Test;
+
+import static org.junit.Assert.*;
+
+public class EndiannessAccessChecks {
+	
+	@Test
+	public void testHeapSegment() {
+		try {
+			testBigAndLittleEndianAccessUnaligned(new HeapMemorySegment(new byte[11111]));
+		}
+		catch (Exception e) {
+			e.printStackTrace();
+			fail(e.getMessage());
+		}
+	}
+
+	@Test
+	public void testHybridOnHeapSegment() {
+		try {
+			testBigAndLittleEndianAccessUnaligned(new HybridMemorySegment(new byte[11111]));
+		}
+		catch (Exception e) {
+			e.printStackTrace();
+			fail(e.getMessage());
+		}
+	}
+
+	@Test
+	public void testHybridOffHeapSegment() {
+		try {
+			testBigAndLittleEndianAccessUnaligned(new HybridMemorySegment(ByteBuffer.allocateDirect(11111)));
+		}
+		catch (Exception e) {
+			e.printStackTrace();
+			fail(e.getMessage());
+		}
+	}
+	
+	private void testBigAndLittleEndianAccessUnaligned(MemorySegment segment) {
+		final Random rnd = new Random();
+		
+		// longs
+		{
+			final long seed = rnd.nextLong();
+			
+			rnd.setSeed(seed);
+			for (int i = 0; i < 10000; i++) {
+				long val = rnd.nextLong();
+				int pos = rnd.nextInt(segment.size - 7);
+				
+				segment.putLongLittleEndian(pos, val);
+				long r = segment.getLongBigEndian(pos);
+				assertEquals(val, Long.reverseBytes(r));
+
+				segment.putLongBigEndian(pos, val);
+				r = segment.getLongLittleEndian(pos);
+				assertEquals(val, Long.reverseBytes(r));
+			}
+		}
+
+		// ints
+		{
+			final long seed = rnd.nextLong();
+
+			rnd.setSeed(seed);
+			for (int i = 0; i < 10000; i++) {
+				int val = rnd.nextInt();
+				int pos = rnd.nextInt(segment.size - 3);
+
+				segment.putIntLittleEndian(pos, val);
+				int r = segment.getIntBigEndian(pos);
+				assertEquals(val, Integer.reverseBytes(r));
+
+				segment.putIntBigEndian(pos, val);
+				r = segment.getIntLittleEndian(pos);
+				assertEquals(val, Integer.reverseBytes(r));
+			}
+		}
+
+		// shorts
+		{
+			final long seed = rnd.nextLong();
+
+			rnd.setSeed(seed);
+			for (int i = 0; i < 10000; i++) {
+				short val = (short) rnd.nextInt();
+				int pos = rnd.nextInt(segment.size - 1);
+
+				segment.putShortLittleEndian(pos, val);
+				short r = segment.getShortBigEndian(pos);
+				assertEquals(val, Short.reverseBytes(r));
+
+				segment.putShortBigEndian(pos, val);
+				r = segment.getShortLittleEndian(pos);
+				assertEquals(val, Short.reverseBytes(r));
+			}
+		}
+
+		// chars
+		{
+			final long seed = rnd.nextLong();
+
+			rnd.setSeed(seed);
+			for (int i = 0; i < 10000; i++) {
+				char val = (char) rnd.nextInt();
+				int pos = rnd.nextInt(segment.size - 1);
+
+				segment.putCharLittleEndian(pos, val);
+				char r = segment.getCharBigEndian(pos);
+				assertEquals(val, Character.reverseBytes(r));
+
+				segment.putCharBigEndian(pos, val);
+				r = segment.getCharLittleEndian(pos);
+				assertEquals(val, Character.reverseBytes(r));
+			}
+		}
+
+		// floats
+		{
+			final long seed = rnd.nextLong();
+
+			rnd.setSeed(seed);
+			for (int i = 0; i < 10000; i++) {
+				float val = rnd.nextFloat();
+				int pos = rnd.nextInt(segment.size - 3);
+
+				segment.putFloatLittleEndian(pos, val);
+				float r = segment.getFloatBigEndian(pos);
+				float reversed = Float.intBitsToFloat(Integer.reverseBytes(Float.floatToRawIntBits(r)));
+				assertEquals(val, reversed, 0.0f);
+
+				segment.putFloatBigEndian(pos, val);
+				r = segment.getFloatLittleEndian(pos);
+				reversed = Float.intBitsToFloat(Integer.reverseBytes(Float.floatToRawIntBits(r)));
+				assertEquals(val, reversed, 0.0f);
+			}
+		}
+
+		// doubles
+		{
+			final long seed = rnd.nextLong();
+
+			rnd.setSeed(seed);
+			for (int i = 0; i < 10000; i++) {
+				double val = rnd.nextDouble();
+				int pos = rnd.nextInt(segment.size - 7);
+
+				segment.putDoubleLittleEndian(pos, val);
+				double r = segment.getDoubleBigEndian(pos);
+				double reversed = Double.longBitsToDouble(Long.reverseBytes(Double.doubleToRawLongBits(r)));
+				assertEquals(val, reversed, 0.0f);
+
+				segment.putDoubleBigEndian(pos, val);
+				r = segment.getDoubleLittleEndian(pos);
+				reversed = Double.longBitsToDouble(Long.reverseBytes(Double.doubleToRawLongBits(r)));
+				assertEquals(val, reversed, 0.0f);
+			}
+		}
+	}
+}

flink-core/src/test/java/org/apache/flink/core/memory/HeapMemorySegmentTest.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.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.nio.ByteBuffer;
+
+import static org.junit.Assert.*;
+
+@RunWith(Parameterized.class)
+public class HeapMemorySegmentTest extends MemorySegmentTestBase {
+
+	public HeapMemorySegmentTest(int pageSize) {
+		super(pageSize);
+	}
+
+	@Override
+	MemorySegment createSegment(int size) {
+		return new HeapMemorySegment(new byte[size]);
+	}
+
+	@Override
+	MemorySegment createSegment(int size, Object owner) {
+		return new HeapMemorySegment(new byte[size], owner);
+	}
+	
+	@Test
+	public void testHeapSegmentSpecifics() {
+		try {
+			final byte[] buffer = new byte[411];
+			HeapMemorySegment seg = new HeapMemorySegment(buffer);
+			
+			assertFalse(seg.isFreed());
+			assertFalse(seg.isOffHeap());
+			assertEquals(buffer.length, seg.size());
+			assertTrue(buffer == seg.getArray());
+
+			ByteBuffer buf1 = seg.wrap(1, 2);
+			ByteBuffer buf2 = seg.wrap(3, 4);
+
+			assertTrue(buf1 != buf2);
+			assertEquals(1, buf1.position());
+			assertEquals(3, buf1.limit());
+			assertEquals(3, buf2.position());
+			assertEquals(7, buf2.limit());
+		}
+		catch (Exception e) {
+			e.printStackTrace();
+			fail(e.getMessage());
+		}
+	}
+}

flink-core/src/test/java/org/apache/flink/core/memory/HybridOffHeapMemorySegmentTest.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.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.nio.ByteBuffer;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+@RunWith(Parameterized.class)
+public class HybridOffHeapMemorySegmentTest extends MemorySegmentTestBase {
+
+	public HybridOffHeapMemorySegmentTest(int pageSize) {
+		super(pageSize);
+	}
+
+	@Override
+	MemorySegment createSegment(int size) {
+		return new HybridMemorySegment(ByteBuffer.allocateDirect(size));
+	}
+
+	@Override
+	MemorySegment createSegment(int size, Object owner) {
+		return new HybridMemorySegment(ByteBuffer.allocateDirect(size), owner);
+	}
+
+	@Test
+	public void testHybridHeapSegmentSpecifics() {
+		try {
+			final ByteBuffer buffer = ByteBuffer.allocateDirect(411);
+			HybridMemorySegment seg = new HybridMemorySegment(buffer);
+
+			assertFalse(seg.isFreed());
+			assertTrue(seg.isOffHeap());
+			assertEquals(buffer.capacity(), seg.size());
+			assertTrue(buffer == seg.getOffHeapBuffer());
+
+			try {
+				seg.getArray();
+				fail("should throw an exception");
+			}
+			catch (IllegalStateException e) {
+				// expected
+			}
+
+			ByteBuffer buf1 = seg.wrap(1, 2);
+			ByteBuffer buf2 = seg.wrap(3, 4);
+
+			assertTrue(buf1 != buffer);
+			assertTrue(buf2 != buffer);
+			assertTrue(buf1 != buf2);
+			assertEquals(1, buf1.position());
+			assertEquals(3, buf1.limit());
+			assertEquals(3, buf2.position());
+			assertEquals(7, buf2.limit());
+		}
+		catch (Exception e) {
+			e.printStackTrace();
+			fail(e.getMessage());
+		}
+	}
+}

flink-core/src/test/java/org/apache/flink/core/memory/HybridOnHeapMemorySegmentTest.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.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.nio.ByteBuffer;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+@RunWith(Parameterized.class)
+public class HybridOnHeapMemorySegmentTest extends MemorySegmentTestBase {
+
+	public HybridOnHeapMemorySegmentTest(int pageSize) {
+		super(pageSize);
+	}
+
+	@Override
+	MemorySegment createSegment(int size) {
+		return new HybridMemorySegment(new byte[size]);
+	}
+
+	@Override
+	MemorySegment createSegment(int size, Object owner) {
+		return new HybridMemorySegment(new byte[size], owner);
+	}
+	
+	@Test
+	public void testHybridHeapSegmentSpecifics() {
+		try {
+			final byte[] buffer = new byte[411];
+			HybridMemorySegment seg = new HybridMemorySegment(buffer);
+
+			assertFalse(seg.isFreed());
+			assertFalse(seg.isOffHeap());
+			assertEquals(buffer.length, seg.size());
+			assertTrue(buffer == seg.getArray());
+			
+			try {
+				seg.getOffHeapBuffer();
+				fail("should throw an exception");
+			}
+			catch (IllegalStateException e) {
+				// expected
+			}
+			
+			ByteBuffer buf1 = seg.wrap(1, 2);
+			ByteBuffer buf2 = seg.wrap(3, 4);
+			
+			assertTrue(buf1 != buf2);
+			assertEquals(1, buf1.position());
+			assertEquals(3, buf1.limit());
+			assertEquals(3, buf2.position());
+			assertEquals(7, buf2.limit());
+		}
+		catch (Exception e) {
+			e.printStackTrace();
+			fail(e.getMessage());
+		}
+	}
+}

flink-core/src/test/java/org/apache/flink/core/memory/MemorySegmentChecksTest.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.Test;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.nio.ByteBuffer;
+
+/**
+ * Tests for the sanity checks of the memory segments.
+ */
+public class MemorySegmentChecksTest {
+	
+	@Test(expected = NullPointerException.class)
+	public void testHeapNullBuffer1() {
+		new HeapMemorySegment(null);
+	}
+
+	@Test(expected = NullPointerException.class)
+	public void testHeapNullBuffer2() {
+		new HeapMemorySegment(null, new Object());
+	}
+
+	@Test(expected = NullPointerException.class)
+	public void testHybridHeapNullBuffer1() {
+		new HybridMemorySegment((byte[]) null);
+	}
+
+	@Test(expected = NullPointerException.class)
+	public void testHybridHeapNullBuffer2() {
+		new HybridMemorySegment((byte[]) null, new Object());
+	}
+
+	@Test(expected = NullPointerException.class)
+	public void testHybridOffHeapNullBuffer1() {
+		new HybridMemorySegment((ByteBuffer) null);
+	}
+
+	@Test(expected = NullPointerException.class)
+	public void testHybridOffHeapNullBuffer2() {
+		new HybridMemorySegment((ByteBuffer) null, new Object());
+	}
+
+	@Test(expected = IllegalArgumentException.class)
+	public void testHybridNonDirectBuffer() {
+		new HybridMemorySegment(ByteBuffer.allocate(1024));
+	}
+
+	@Test(expected = IllegalArgumentException.class)
+	public void testZeroAddress(){
+		new MockSegment(0L, 4*1024, null);
+	}
+
+	@Test(expected = IllegalArgumentException.class)
+	public void testNegativeAddress(){
+		new MockSegment(-1L, 4*1024, null);
+	}
+
+	@Test(expected = IllegalArgumentException.class)
+	public void testTooLargeAddress(){
+		new MockSegment(Long.MAX_VALUE - 8*1024, 4*1024, null);
+	}
+	
+	// ------------------------------------------------------------------------
+	
+	final class MockSegment extends MemorySegment {
+
+		MockSegment(long offHeapAddress, int size, Object owner) {
+			super(offHeapAddress, size, owner);
+		}
+
+		@Override
+		public ByteBuffer wrap(int offset, int length) {
+			return null;
+		}
+
+		@Override
+		public byte get(int index) {
+			return 0;
+		}
+
+		@Override
+		public void put(int index, byte b) {}
+
+		@Override
+		public void get(int index, byte[] dst) {}
+
+		@Override
+		public void put(int index, byte[] src) {}
+
+		@Override
+		public void get(int index, byte[] dst, int offset, int length) {}
+
+		@Override
+		public void put(int index, byte[] src, int offset, int length) {}
+
+		@Override
+		public boolean getBoolean(int index) {
+			return false;
+		}
+
+		@Override
+		public void putBoolean(int index, boolean value) {}
+
+		@Override
+		public void get(DataOutput out, int offset, int length) {}
+
+		@Override
+		public void put(DataInput in, int offset, int length) {}
+
+		@Override
+		public void get(int offset, ByteBuffer target, int numBytes) {}
+
+		@Override
+		public void put(int offset, ByteBuffer source, int numBytes) {}
+	};
+}

flink-core/src/test/java/org/apache/flink/types/NormalizableKeyTest.java

@@ -16,18 +16,12 @@
  * limitations under the License.
  */
 
-
 package org.apache.flink.types;
 
-import org.junit.Assert;
-
 import org.apache.flink.core.memory.MemorySegment;
-import org.apache.flink.types.CharValue;
-import org.apache.flink.types.IntValue;
-import org.apache.flink.types.LongValue;
-import org.apache.flink.types.NormalizableKey;
-import org.apache.flink.types.NullValue;
-import org.apache.flink.types.StringValue;
+import org.apache.flink.core.memory.MemorySegmentFactory;
+
+import org.junit.Assert;
 import org.junit.Test;
 
 public class NormalizableKeyTest {
@@ -124,9 +118,9 @@ public class NormalizableKeyTest {
 		
 		for (int i = 0; i < 5; i++) {
 			// self checks
-			for (int x = 0; x < allChars.length; x++) {
-				for (int y = 0; y < allChars.length; y++) {
-					assertNormalizableKey(allChars[x], allChars[y], i);
+			for (CharValue allChar1 : allChars) {
+				for (CharValue allChar : allChars) {
+					assertNormalizableKey(allChar1, allChar, i);
 				}
 			}
 		}
@@ -135,8 +129,8 @@ public class NormalizableKeyTest {
 	@SuppressWarnings("unchecked")
 	private <T extends Comparable<T>> void assertNormalizableKey(NormalizableKey<T> key1, NormalizableKey<T> key2, int len) {
 		
-		byte[] normalizedKeys = new byte[2*len];
-		MemorySegment wrapper = new MemorySegment(normalizedKeys);
+		byte[] normalizedKeys = new byte[32];
+		MemorySegment wrapper = MemorySegmentFactory.wrap(normalizedKeys);
 		
 		key1.copyNormalizedKey(wrapper, 0, len);
 		key2.copyNormalizedKey(wrapper, len, len);
@@ -147,13 +141,13 @@ public class NormalizableKeyTest {
 			int normKey2 = normalizedKeys[len + i] & 0xFF;
 			
 			if ((comp = (normKey1 - normKey2)) != 0) {
-				if (Math.signum(((T) key1).compareTo((T) key2)) != Math.signum(comp)) {
+				if (Math.signum(key1.compareTo((T) key2)) != Math.signum(comp)) {
 					Assert.fail("Normalized key comparison differs from actual key comparision");
 				}
 				return;
 			}
 		}
-		if (((T) key1).compareTo((T) key2) != 0 && key1.getMaxNormalizedKeyLen() <= len) {
+		if (key1.compareTo((T) key2) != 0 && key1.getMaxNormalizedKeyLen() <= len) {
 			Assert.fail("Normalized key was not able to distinguish keys, " +
 					"although it should as the length of it sufficies to uniquely identify them");
 		}

flink-runtime/src/main/java/org/apache/flink/runtime/execution/Environment.java

@@ -29,7 +29,7 @@ import org.apache.flink.runtime.io.network.partition.consumer.InputGate;
 import org.apache.flink.api.common.JobID;
 import org.apache.flink.runtime.jobgraph.JobVertexID;
 import org.apache.flink.runtime.jobgraph.tasks.InputSplitProvider;
-import org.apache.flink.runtime.memorymanager.MemoryManager;
+import org.apache.flink.runtime.memory.MemoryManager;
 import org.apache.flink.runtime.state.StateHandle;
 import org.apache.flink.runtime.taskmanager.TaskManagerRuntimeInfo;
 

flink-runtime/src/main/java/org/apache/flink/runtime/io/disk/FileChannelInputView.java

@@ -27,8 +27,8 @@ import java.util.List;
 
 import org.apache.flink.core.memory.MemorySegment;
 import org.apache.flink.runtime.io.disk.iomanager.BlockChannelReader;
-import org.apache.flink.runtime.memorymanager.AbstractPagedInputView;
-import org.apache.flink.runtime.memorymanager.MemoryManager;
+import org.apache.flink.runtime.memory.AbstractPagedInputView;
+import org.apache.flink.runtime.memory.MemoryManager;
 import org.apache.flink.runtime.util.MathUtils;
 
 /**

flink-runtime/src/main/java/org/apache/flink/runtime/io/disk/FileChannelOutputView.java

@@ -26,8 +26,8 @@ import java.util.List;
 
 import org.apache.flink.core.memory.MemorySegment;
 import org.apache.flink.runtime.io.disk.iomanager.BlockChannelWriter;
-import org.apache.flink.runtime.memorymanager.AbstractPagedOutputView;
-import org.apache.flink.runtime.memorymanager.MemoryManager;
+import org.apache.flink.runtime.memory.AbstractPagedOutputView;
+import org.apache.flink.runtime.memory.MemoryManager;
 
 /**
  * A {@link org.apache.flink.core.memory.DataOutputView} that is backed by a {@link BlockChannelWriter}, making it effectively a data output

flink-runtime/src/main/java/org/apache/flink/runtime/io/disk/RandomAccessInputView.java

@@ -23,7 +23,7 @@ import java.util.ArrayList;
 
 import org.apache.flink.core.memory.MemorySegment;
 import org.apache.flink.core.memory.SeekableDataInputView;
-import org.apache.flink.runtime.memorymanager.AbstractPagedInputView;
+import org.apache.flink.runtime.memory.AbstractPagedInputView;
 import org.apache.flink.runtime.util.MathUtils;
 
 

flink-runtime/src/main/java/org/apache/flink/runtime/io/disk/RandomAccessOutputView.java

@@ -23,7 +23,7 @@ import java.io.EOFException;
 
 import org.apache.flink.core.memory.MemorySegment;
 import org.apache.flink.core.memory.SeekableDataOutputView;
-import org.apache.flink.runtime.memorymanager.AbstractPagedOutputView;
+import org.apache.flink.runtime.memory.AbstractPagedOutputView;
 import org.apache.flink.runtime.util.MathUtils;
 
 

flink-runtime/src/main/java/org/apache/flink/runtime/io/disk/SeekableFileChannelInputView.java

@@ -28,8 +28,8 @@ import org.apache.flink.core.memory.MemorySegment;
 import org.apache.flink.runtime.io.disk.iomanager.BlockChannelReader;
 import org.apache.flink.runtime.io.disk.iomanager.FileIOChannel;
 import org.apache.flink.runtime.io.disk.iomanager.IOManager;
-import org.apache.flink.runtime.memorymanager.AbstractPagedInputView;
-import org.apache.flink.runtime.memorymanager.MemoryManager;
+import org.apache.flink.runtime.memory.AbstractPagedInputView;
+import org.apache.flink.runtime.memory.MemoryManager;
 import org.apache.flink.runtime.util.MathUtils;
 
 /**

flink-runtime/src/main/java/org/apache/flink/runtime/io/disk/SimpleCollectingOutputView.java

@@ -25,7 +25,7 @@ import java.util.List;
 
 import org.apache.flink.core.memory.MemorySegment;
 import org.apache.flink.core.memory.MemorySegmentSource;
-import org.apache.flink.runtime.memorymanager.AbstractPagedOutputView;
+import org.apache.flink.runtime.memory.AbstractPagedOutputView;
 import org.apache.flink.runtime.util.MathUtils;
 
 /**

flink-runtime/src/main/java/org/apache/flink/runtime/io/disk/SpillingBuffer.java

@@ -30,7 +30,7 @@ import org.apache.flink.runtime.io.disk.iomanager.BlockChannelReader;
 import org.apache.flink.runtime.io.disk.iomanager.BlockChannelWriter;
 import org.apache.flink.runtime.io.disk.iomanager.HeaderlessChannelReaderInputView;
 import org.apache.flink.runtime.io.disk.iomanager.IOManager;
-import org.apache.flink.runtime.memorymanager.AbstractPagedOutputView;
+import org.apache.flink.runtime.memory.AbstractPagedOutputView;
 
 
 /**

flink-runtime/src/main/java/org/apache/flink/runtime/io/disk/iomanager/ChannelReaderInputView.java

@@ -25,7 +25,7 @@ import java.util.List;
 import java.util.concurrent.LinkedBlockingQueue;
 
 import org.apache.flink.core.memory.MemorySegment;
-import org.apache.flink.runtime.memorymanager.AbstractPagedInputView;
+import org.apache.flink.runtime.memory.AbstractPagedInputView;
 
 /**
  * A {@link org.apache.flink.core.memory.DataInputView} that is backed by a

flink-runtime/src/main/java/org/apache/flink/runtime/io/disk/iomanager/ChannelWriterOutputView.java

@@ -24,7 +24,7 @@ import java.util.List;
 import java.util.concurrent.LinkedBlockingQueue;
 
 import org.apache.flink.core.memory.MemorySegment;
-import org.apache.flink.runtime.memorymanager.AbstractPagedOutputView;
+import org.apache.flink.runtime.memory.AbstractPagedOutputView;
 
 /**
  * A {@link org.apache.flink.core.memory.DataOutputView} that is backed by a

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

@@ -107,7 +107,8 @@ public class NetworkEnvironment {
 		// create the network buffers - this is the operation most likely to fail upon
 		// mis-configuration, so we do this first
 		try {
-			networkBufferPool = new NetworkBufferPool(config.numNetworkBuffers(), config.networkBufferSize());
+			networkBufferPool = new NetworkBufferPool(config.numNetworkBuffers(),
+					config.networkBufferSize(), config.memoryType());
 		}
 		catch (Throwable t) {
 			throw new IOException("Cannot allocate network buffer pool: " + t.getMessage(), t);

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

@@ -230,21 +230,21 @@ public class AdaptiveSpanningRecordDeserializer<T extends IOReadableWritable> im
 
 		@Override
 		public final short readShort() throws IOException {
-			final short v = this.segment.getShort(this.position);
+			final short v = this.segment.getShortBigEndian(this.position);
 			this.position += 2;
 			return v;
 		}
 
 		@Override
 		public final int readUnsignedShort() throws IOException {
-			final int v = this.segment.getShort(this.position) & 0xffff;
+			final int v = this.segment.getShortBigEndian(this.position) & 0xffff;
 			this.position += 2;
 			return v;
 		}
 
 		@Override
 		public final char readChar() throws IOException  {
-			final char v = this.segment.getChar(this.position);
+			final char v = this.segment.getCharBigEndian(this.position);
 			this.position += 2;
 			return v;
 		}

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

@@ -19,6 +19,7 @@
 package org.apache.flink.runtime.io.network.api.serialization;
 
 import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.core.memory.MemorySegmentFactory;
 import org.apache.flink.runtime.event.AbstractEvent;
 import org.apache.flink.runtime.io.network.api.CheckpointBarrier;
 import org.apache.flink.runtime.io.network.api.EndOfPartitionEvent;
@@ -145,8 +146,9 @@ public class EventSerializer {
 	public static Buffer toBuffer(AbstractEvent event) {
 		final ByteBuffer serializedEvent = EventSerializer.toSerializedEvent(event);
 
-		final Buffer buffer = new Buffer(new MemorySegment(serializedEvent.array()),
-				FreeingBufferRecycler.INSTANCE, false);
+		MemorySegment data = MemorySegmentFactory.wrap(serializedEvent.array());
+		
+		final Buffer buffer = new Buffer(data, FreeingBufferRecycler.INSTANCE, false);
 		buffer.setSize(serializedEvent.remaining());
 
 		return buffer;