[runtime] [tests] Fix possible NPE and add Netty serialization test

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

@@ -147,8 +147,11 @@ abstract class NettyMessage {
 			else if (msgId == ErrorResponse.ID) {
 				decodedMsg = new ErrorResponse();
 			}
+			else if (msgId == CancelPartitionRequest.ID) {
+				decodedMsg = new CancelPartitionRequest();
+			}
 			else {
-				throw new IllegalStateException("Received unknown message from producer: " + decodedMsg.getClass());
+				throw new IllegalStateException("Received unknown message from producer: " + msg);
 			}
 
 			if (decodedMsg != null) {
@@ -486,6 +489,9 @@ abstract class NettyMessage {
 
 		InputChannelID receiverId;
 
+		public CancelPartitionRequest() {
+		}
+
 		public CancelPartitionRequest(InputChannelID receiverId) {
 			this.receiverId = receiverId;
 		}
@@ -495,7 +501,7 @@ abstract class NettyMessage {
 			ByteBuf result = null;
 
 			try {
-				result = allocateBuffer(allocator, ID);
+				result = allocateBuffer(allocator, ID, 16);
 				receiverId.writeTo(result);
 			}
 			catch (Throwable t) {

flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/NettyMessageSerializationTest.java

@@ -140,6 +140,13 @@ public class NettyMessageSerializationTest {
 			assertEquals(expected.partitionId, actual.partitionId);
 			assertEquals(expected.receiverId, actual.receiverId);
 		}
+
+		{
+			NettyMessage.CancelPartitionRequest expected = new NettyMessage.CancelPartitionRequest(new InputChannelID());
+			NettyMessage.CancelPartitionRequest actual = encodeAndDecode(expected);
+
+			assertEquals(expected.receiverId, actual.receiverId);
+		}
 	}
 
 	@SuppressWarnings("unchecked")