Kafka spout PartitionManager should emit failed tuples without replaying all data that follows

[AdJugglerDan]

PartitionManager resets to the offset of a failed message and upon nextTuple() starts re-transmitting all data from that offset forward. We modified this behavior to cache the data emitted and only re-transmit the failed data on nextTuple(). I would like to submit the changes for review/inclusion if others feel that this is worthwhile. The following changes were made:

*Add a HashMap<Long, List> for pending messages
*Add a LinkedList for offsets of failed messages
*Change fail(Long offset) to only add the offset to the failed offset list
*Change ack(Long offset) to also remove the entry of the acked offset from the HashMap
*Change next(SpoutOutputCollector collector) to re-emit failed messages if list is not empty

Respectfully Submitted