commit 8d4c9032858b1f57ee8835d5169ab7d964698a45
author Neng Lu <freeneng@gmail.com> Mon Jul 17 16:16:55 2017 -0700
committer Huijun Wu <huijunw@twitter.com> Mon Aug 07 12:46:56 2017 -0700
tree fbd2d8128d5ac1ab727c2f3ac5a7cadc5a7f6718
parent e7c5f829412020a517488a7a5ddef1e0920245f0

fix kafka spout HeapByteBuffer not serializable problem (#2063)

heron/storm/src/java/org/apache/storm/spout/RawMultiScheme.java

diff --git a/heron/storm/src/java/org/apache/storm/spout/RawMultiScheme.java b/heron/storm/src/java/org/apache/storm/spout/RawMultiScheme.java
index 620fc87..c1aaa89 100644
--- a/heron/storm/src/java/org/apache/storm/spout/RawMultiScheme.java
+++ b/heron/storm/src/java/org/apache/storm/spout/RawMultiScheme.java
@@ -31,7 +31,9 @@
 
   @Override
   public Iterable<List<Object>> deserialize(ByteBuffer ser) {
-    return asList(tuple(ser));
+    byte[] bytes = new byte[ser.remaining()];
+    ser.get(bytes);
+    return asList(tuple(bytes));
   }
 
   @Override

heron/storm/src/java/org/apache/storm/spout/RawScheme.java

diff --git a/heron/storm/src/java/org/apache/storm/spout/RawScheme.java b/heron/storm/src/java/org/apache/storm/spout/RawScheme.java
index e36add4..30755b6 100644
--- a/heron/storm/src/java/org/apache/storm/spout/RawScheme.java
+++ b/heron/storm/src/java/org/apache/storm/spout/RawScheme.java
@@ -28,10 +28,14 @@
 public class RawScheme implements Scheme {
   private static final long serialVersionUID = 6098042939916415521L;
 
+  @Override
   public List<Object> deserialize(ByteBuffer ser) {
-    return tuple(ser);
+    byte[] bytes = new byte[ser.remaining()];
+    ser.get(bytes);
+    return tuple(bytes);
   }
 
+  @Override
   public Fields getOutputFields() {
     return new Fields("bytes");
   }