fixed bug handling incomplete messages at end of batch; resume consumer when the current kafka segment is deleted prematurely

Author: quipo

README.md

@@ -8,6 +8,11 @@ Go language: (http://golang.org/) <br/>
 
 ## Changes
 
+### May 2015
+
+* fixed bug handling partial message at end of a fetch response when the payload is < 4 bytes
+* if the the kafka log segment being read is cleaned up, attempt resuming the consumer from the earliest available offset
+
 ### April 2015
 
 * added support for Snappy compression

consumer.go

@@ -92,7 +92,7 @@ func (consumer *BrokerConsumer) ConsumeOnChannel(msgChan chan *Message, pollTime
 			}, quit)
 
 			if err != nil {
-				if err != io.EOF && err.Error() != "use of closed network connection" { //
+				if err != io.EOF && err.Error() != "use of closed network connection" {
 					log.Println("Fatal Error: ", err)
 					panic(err)
 				}
@@ -129,6 +129,23 @@ func (consumer *BrokerConsumer) ConsumeOnChannel(msgChan chan *Message, pollTime
 // MessageHandlerFunc defines the interface for message handlers accepted by Consume()
 type MessageHandlerFunc func(msg *Message)
 
+// reconnect from the earliest available offset after the current one becomes unavailable
+func (consumer *BrokerConsumer) reconnectFromEarliestAvailableOffset(conn *net.TCPConn) (uint64, error) {
+	_, err := conn.Write(consumer.broker.EncodeOffsetRequest(OFFSET_EARLIEST, 1))
+	if err != nil {
+		log.Println("Failed kafka offset request:", err.Error())
+		return 0, err
+	}
+
+	length, payload, err := consumer.broker.readResponse(conn)
+	log.Println("kafka offset request of", length, "bytes starting from offset", consumer.offset, payload)
+
+	if err != nil {
+		return 0, err
+	}
+	return binary.BigEndian.Uint64(payload[0:]), nil
+}
+
 // Consume makes a single fetch request and sends the messages in the message set to a handler function
 func (consumer *BrokerConsumer) Consume(handlerFunc MessageHandlerFunc, stop <-chan struct{}) (int, error) {
 	conn, err := consumer.broker.connect()
@@ -157,7 +174,25 @@ func (consumer *BrokerConsumer) consumeWithConn(conn *net.TCPConn, handlerFunc M
 	//log.Println("kafka fetch request of", length, "bytes starting from offset", consumer.offset)
 
 	if err != nil {
-		return -1, err
+		if err.Error() != "Broker Response Error: 1" {
+			return -1, err
+		}
+
+		// special case: reset offset if kafka cleaned up the file being read
+		log.Println("ERROR fetching kafka batch at offset", consumer.offset, "- probably due to timeout or premature cleanup of kafka data file")
+		log.Println("Fetching earliest available offset in kafka log")
+		consumer.offset, err = consumer.reconnectFromEarliestAvailableOffset(conn)
+		if err != nil {
+			panic(err)
+		}
+		log.Println("Resuming at offset", consumer.offset)
+		length, payload, err = consumer.broker.readResponse(conn)
+		if err != nil {
+			log.Println("Cannot resume consuming at new offset, needs manual intervention")
+			if err.Error() != "Broker Response Error: 1" {
+				return -1, err
+			}
+		}
 	}
 
 	num := 0
@@ -166,14 +201,15 @@ func (consumer *BrokerConsumer) consumeWithConn(conn *net.TCPConn, handlerFunc M
 		currentOffset := uint64(0)
 		for currentOffset <= uint64(length-4) {
 			totalLength, msgs, err1 := Decode(payload[currentOffset:], consumer.codecs)
-			if ErrIncompletePacket == err1 {
+			if ErrIncompletePacket == err1 || ErrMalformedPacket == err1 {
 				// Reached the end of the current packet and the last message is incomplete.
 				if 0 == num {
 					// This is the very first message in the batch => we need to request a larger packet
 					// or the consumer will get stuck here indefinitely
 					log.Printf("ERROR: Incomplete message at offset %d %d, change the configuration to a larger max fetch size\n",
 						consumer.offset,
 						currentOffset)
+					log.Printf("\nPayload length: %d, currentOffset: %d, payload: [%x]\n\n", length, currentOffset, payload)
 				} else {
 					// Partial message at end of current batch, need a new Fetch Request from a newer offset
 					log.Printf("DEBUG: Incomplete message at offset %d %d for topic '%s' (%s, partition %d), fetching new batch from offset %d\n",
@@ -185,10 +221,6 @@ func (consumer *BrokerConsumer) consumeWithConn(conn *net.TCPConn, handlerFunc M
 						consumer.offset+currentOffset)
 				}
 				break
-			} else if ErrMalformedPacket == err1 {
-				log.Printf("ERROR: Malformed message at offset %d %d\n",
-					consumer.offset,
-					currentOffset)
 			}
 			if err != nil {
 				log.Printf("Payload length: %d, currentOffset: %d, payload: [%x]", length, currentOffset, payload)

kafka.go

@@ -44,9 +44,11 @@ type Broker struct {
 }
 
 func newBroker(hostname string, topic string, partition int) *Broker {
-	return &Broker{topic: topic,
+	return &Broker{
+		topic:     topic,
 		partition: partition,
-		hostname:  hostname}
+		hostname:  hostname,
+	}
 }
 
 func (b *Broker) connect() (conn *net.TCPConn, error error) {

kafka_test.go

@@ -230,7 +230,8 @@ func TestLongCompressedMessageRoundTripSnappy(t *testing.T) {
 }
 
 func TestMultipleCompressedMessages(t *testing.T) {
-	msgs := []*Message{NewMessage([]byte("testing")),
+	msgs := []*Message{
+		NewMessage([]byte("testing")),
 		NewMessage([]byte("multiple")),
 		NewMessage([]byte("messages")),
 	}

message.go

@@ -150,14 +150,18 @@ func Decode(packet []byte, payloadCodecsMap map[byte]PayloadCodec) (uint32, []Me
 		start := payloadLen - messageLenLeft
 		innerMsg, err = decodeMessage(message.payload[start:], payloadCodecsMap)
 		if nil != err {
-			log.Println("DEBUG:", err.Error())
+			log.Println("DEBUG: (inner compressed msg)", err.Error())
 			if ErrIncompletePacket == err {
-				// the current top-level message is incomplete, reached end of packet
-				err = nil
+				// the current top-level message is incomplete, reached end of packet:
+				// we need a larger packet. Given we cannot advance the cursor within
+				// a compressed message, we discard all the inner messages read so far
+				// and force a new request. The checksum on the first decodeMessage() call
+				// in this function should had prevented reaching this point though...
+				return 0, []Message{}, err
 			}
 			break
 		}
-		messageLenLeft = messageLenLeft - innerMsg.totalLength - 4 // message length uint32
+		messageLenLeft -= (4 + innerMsg.totalLength) // message length uint32
 		messages = append(messages, *innerMsg)
 	}
 

request.go

@@ -37,6 +37,8 @@ const (
 	REQUEST_MULTIFETCH               = 2
 	REQUEST_MULTIPRODUCE             = 3
 	REQUEST_OFFSETS                  = 4
+	OFFSET_LATEST        int64       = -1
+	OFFSET_EARLIEST      int64       = -2
 )
 
 // EncodeRequestHeader marshals a request into kafka's wire format