package com.hpcloud.mon.infrastructure.thresholding;

import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Values;

import com.hpcloud.mon.EventSpoutConfig;
import com.hpcloud.mon.infrastructure.thresholding.deserializer.EventDeserializer;

public class EventSpout extends KafkaSpout {
    private static final Logger LOG = LoggerFactory.getLogger(EventSpout.class);

    private static final long serialVersionUID = 8457340455857276878L;

    private final EventDeserializer deserializer;

    public EventSpout(EventSpoutConfig configuration, EventDeserializer deserializer) {
        super(configuration.kafkaConsumerConfiguration);
        this.deserializer = deserializer;
        LOG.info("EventSpout created");
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(deserializer.getOutputFields());
    }

    @Override
    protected void processMessage(byte[] message, SpoutOutputCollector collector) {
        List<List<?>> events = deserializer.deserialize(message);
        if (events != null) {
            for (final List<?> event : events) {
                collector.emit(new Values(event.get(0)));
            }
        }
    }
}

@@ -47,16 +47,45 @@
 lding;%0A%0A
+import java.io.Serializable;%0A
 import j
@@ -1294,24 +1294,295 @@
 : events) %7B%0A
+                final Object eventToSend = event.get(0);%0A                if (!(eventToSend instanceof Serializable)) %7B%0A                    LOG.error(%22Class %7B%7D is not Serializable: %7B%7D%22, eventToSend.getClass(), eventToSend);%0A                    continue;%0A                %7D%0A
             
@@ -1616,23 +1616,22 @@
 es(event
-.get(0)
+ToSend
 ));%0A    

Check that everything coming in on the EventSpout is Serializable before sending it out so the Spout doesn't die.