Merge pull request #119 from Shopify/validate-offset

Validate the offset received from kafka when consuming

consumer.go

@@ -265,13 +265,23 @@ func (c *Consumer) fetchMessages() {
 
 		for _, msgBlock := range block.MsgSet.Messages {
 			for _, msg := range msgBlock.Messages() {
+
+				event := &ConsumerEvent{Topic: c.topic, Partition: c.partition}
+				if msg.Offset != c.offset {
+					event.Err = IncompleteResponse
+				} else {
+					event.Key = msg.Msg.Key
+					event.Value = msg.Msg.Value
+					event.Offset = msg.Offset
+					c.offset++
+				}
+
 				select {
 				case <-c.stopper:
 					close(c.events)
 					close(c.done)
 					return
-				case c.events <- &ConsumerEvent{Key: msg.Msg.Key, Value: msg.Msg.Value, Offset: msg.Offset, Topic: c.topic, Partition: c.partition}:
-					c.offset++
+				case c.events <- event:
 				}
 			}
 		}