title: NullPointerException in storm when deserialize tuples
body: <p>I've encountered the following NPE when storm tries to deserialize. I did not use OutputCollector concurrently in my code. The only object we are passing between bolts are a thrift object, and we have written a serializer for it. I've attached the code of serializer and please help to check whether there are any potential bugs there.</p>
  
  <pre><code>2016-03-04 17:17:43.583 b.s.util [ERROR] Async loop died!
  java.lang.RuntimeException: java.lang.NullPointerException
          at backtype.storm.utils.DisruptorQueue.consumeBatchToCursor(DisruptorQueue.java:135) ~[storm-core-0.10.0.jar:0.10.0]
          at backtype.storm.utils.DisruptorQueue.consumeBatchWhenAvailable(DisruptorQueue.java:106) ~[storm-core-0.10.0.jar:0.10.0]
          at backtype.storm.disruptor$consume_batch_when_available.invoke(disruptor.clj:80) ~[storm-core-0.10.0.jar:0.10.0]
          at backtype.storm.daemon.executor$fn__5694$fn__5707$fn__5758.invoke(executor.clj:819) ~[storm-core-0.10.0.jar:0.10.0]
          at backtype.storm.util$async_loop$fn__545.invoke(util.clj:479) [storm-core-0.10.0.jar:0.10.0]
          at clojure.lang.AFn.run(AFn.java:22) [clojure-1.6.0.jar:?]
          at java.lang.Thread.run(Thread.java:745) [?:1.8.0_60]
  Caused by: java.lang.NullPointerException
          at com.esotericsoftware.kryo.io.Input.setBuffer(Input.java:57) ~[kryo-2.21.jar:?]
          at backtype.storm.serialization.KryoTupleDeserializer.deserialize(KryoTupleDeserializer.java:47) ~[storm-core-0.10.0.jar:0.10.0]
          at backtype.storm.daemon.executor$mk_task_receiver$fn__5615.invoke(executor.clj:433) ~[storm-core-0.10.0.jar:0.10.0]
          at backtype.storm.disruptor$clojure_handler$reify__5189.onEvent(disruptor.clj:58) ~[storm-core-0.10.0.jar:0.10.0]
          at backtype.storm.utils.DisruptorQueue.consumeBatchToCursor(DisruptorQueue.java:132) ~[storm-core-0.10.0.jar:0.10.0]
          ... 6 more
  2016-03-04 17:17:43.584 b.s.d.executor [ERROR]
  java.lang.RuntimeException: java.lang.NullPointerException
          at backtype.storm.utils.DisruptorQueue.consumeBatchToCursor(DisruptorQueue.java:135) ~[storm-core-0.10.0.jar:0.10.0]
          at backtype.storm.utils.DisruptorQueue.consumeBatchWhenAvailable(DisruptorQueue.java:106) ~[storm-core-0.10.0.jar:0.10.0]
          at backtype.storm.disruptor$consume_batch_when_available.invoke(disruptor.clj:80) ~[storm-core-0.10.0.jar:0.10.0]
          at backtype.storm.daemon.executor$fn__5694$fn__5707$fn__5758.invoke(executor.clj:819) ~[storm-core-0.10.0.jar:0.10.0]
          at backtype.storm.util$async_loop$fn__545.invoke(util.clj:479) [storm-core-0.10.0.jar:0.10.0]
          at clojure.lang.AFn.run(AFn.java:22) [clojure-1.6.0.jar:?]
          at java.lang.Thread.run(Thread.java:745) [?:1.8.0_60]
  Caused by: java.lang.NullPointerException
          at com.esotericsoftware.kryo.io.Input.setBuffer(Input.java:57) ~[kryo-2.21.jar:?]
          at backtype.storm.serialization.KryoTupleDeserializer.deserialize(KryoTupleDeserializer.java:47) ~[storm-core-0.10.0.jar:0.10.0]
          at backtype.storm.daemon.executor$mk_task_receiver$fn__5615.invoke(executor.clj:433) ~[storm-core-0.10.0.jar:0.10.0]
          at backtype.storm.disruptor$clojure_handler$reify__5189.onEvent(disruptor.clj:58) ~[storm-core-0.10.0.jar:0.10.0]
          at backtype.storm.utils.DisruptorQueue.consumeBatchToCursor(DisruptorQueue.java:132) ~[storm-core-0.10.0.jar:0.10.0]
          ... 6 more
  2016-03-04 17:17:43.648 b.s.util [ERROR] Halting process: ("Worker died")
  java.lang.RuntimeException: ("Worker died")
          at backtype.storm.util$exit_process_BANG_.doInvoke(util.clj:336) [storm-core-0.10.0.jar:0.10.0]
          at clojure.lang.RestFn.invoke(RestFn.java:423) [clojure-1.6.0.jar:?]
          at backtype.storm.daemon.worker$fn__7188$fn__7189.invoke(worker.clj:536) [storm-core-0.10.0.jar:0.10.0]
          at backtype.storm.daemon.executor$mk_executor_data$fn__5523$fn__5524.invoke(executor.clj:261) [storm-core-0.10.0.jar:0.10.0]
          at backtype.storm.util$async_loop$fn__545.invoke(util.clj:489) [storm-core-0.10.0.jar:0.10.0]
          at clojure.lang.AFn.run(AFn.java:22) [clojure-1.6.0.jar:?]
          at java.lang.Thread.run(Thread.java:745) [?:1.8.0_60]
  </code></pre>
  
  <hr>
  
  <pre><code>import com.esotericsoftware.kryo.Kryo;
  import com.esotericsoftware.kryo.Serializer;
  import com.esotericsoftware.kryo.io.Input;
  import com.esotericsoftware.kryo.io.Output;
  import org.apache.thrift.TBase;
  import org.apache.thrift.TDeserializer;
  import org.apache.thrift.TException;
  import org.apache.thrift.TSerializer;
  import org.apache.thrift.protocol.TCompactProtocol;
  import org.slf4j.Logger;
  
  /**
   * @author zhangyf
   * @version 1.0 2016-02-19
   */
  
  public class GZipThriftSerializer&lt;T extends TBase&gt; extends Serializer&lt;T&gt; {
  
    private static final int EMPTY = 0;
  
    // add DELTA to make length positive due to kryo internal optimization
    private static final int DELTA = 1;
  
    private static final Logger LOG = com.mediav.utils.Loggers.get();
  
    private static final ThreadLocal&lt;TSerializer&gt; COMPACT_T_SERIALIZER = new ThreadLocal&lt;TSerializer&gt;() {
      @Override
      protected TSerializer initialValue() {
        return new TSerializer(new TCompactProtocol.Factory());
      }
    };
  
    private static final ThreadLocal&lt;TDeserializer&gt; COMPACT_T_DESERIALIZER = new ThreadLocal&lt;TDeserializer&gt;() {
      @Override
      protected TDeserializer initialValue() {
        return new TDeserializer(new TCompactProtocol.Factory());
      }
    };
  
    {
      setAcceptsNull(true);
    }
  
    @Override
    public void write(Kryo kryo, Output output, T t) {
      if (t == null) {
        output.writeInt(EMPTY + DELTA, true);
        return;
      }
      try {
        byte[] data = COMPACT_T_SERIALIZER.get().serialize(t);
        byte[] compressed = GZipUtils.zip(data);
        output.writeInt(compressed.length + DELTA, true);
        output.writeBytes(compressed);
      } catch (TException e) {
        LOG.error("TException during serialization", e);
        output.writeInt(EMPTY + DELTA, true);
      }
    }
  
    @Override
    public T read(Kryo kryo, Input input, Class&lt;T&gt; aClass) {
      int length = input.readInt(true) - DELTA;
      if (length == 0) {
        return null;
      }
      try {
        T t = aClass.newInstance();
        COMPACT_T_DESERIALIZER.get().deserialize(t, GZipUtils.unzip(input.readBytes(length)));
        return t;
      } catch (InstantiationException e) {
        throw new RuntimeException(e);
      } catch (IllegalAccessException e) {
        throw new RuntimeException(e);
      } catch (TException e) {
        LOG.error("TException during deserialization", e);
      }
      return null;
    }
  }
  </code></pre>