flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From u..@apache.org
Subject flink git commit: [runtime] [tests] Fix possible NPE and add Netty serialization test
Date Tue, 26 May 2015 09:49:47 GMT
Repository: flink
Updated Branches:
  refs/heads/master ae446388b -> e0d1fd504


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


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/e0d1fd50
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/e0d1fd50
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/e0d1fd50

Branch: refs/heads/master
Commit: e0d1fd504d6313b1b02a06ef7935cf7fd7069a39
Parents: ae44638
Author: Ufuk Celebi <uce@apache.org>
Authored: Tue May 26 11:49:32 2015 +0200
Committer: Ufuk Celebi <uce@apache.org>
Committed: Tue May 26 11:49:44 2015 +0200

----------------------------------------------------------------------
 .../flink/runtime/io/network/netty/NettyMessage.java      | 10 ++++++++--
 .../io/network/netty/NettyMessageSerializationTest.java   |  7 +++++++
 2 files changed, 15 insertions(+), 2 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/e0d1fd50/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyMessage.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyMessage.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyMessage.java
index db5c542..d0840b0 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyMessage.java
+++ b/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) {

http://git-wip-us.apache.org/repos/asf/flink/blob/e0d1fd50/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/NettyMessageSerializationTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/NettyMessageSerializationTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/NettyMessageSerializationTest.java
index b2fe8a9..b464692 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/NettyMessageSerializationTest.java
+++ b/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")


Mime
View raw message