Account for possible offset skew

consumer.go

@@ -524,10 +524,20 @@ func (child *partitionConsumer) parseRecords(batch *RecordBatch) ([]*ConsumerMes
 	var messages []*ConsumerMessage
 	var incomplete bool
 	prelude := true
-	originalOffset := child.offset
+
+	// Compacted records can lie about their offset delta, but that can be
+	// detected comparing the last offset delta from the record batch with
+	// actual offset delta of the last record. The difference (skew) should be
+	// applied to all records in the batch.
+	var offsetSkew int64
+	recordCount := len(batch.Records)
+	if recordCount > 0 {
+		lastRecordOffsetDelta := batch.Records[recordCount-1].OffsetDelta
+		offsetSkew = int64(batch.LastOffsetDelta) - lastRecordOffsetDelta
+	}
 
 	for _, rec := range batch.Records {
-		offset := batch.FirstOffset + rec.OffsetDelta
+		offset := batch.FirstOffset + rec.OffsetDelta + offsetSkew
 		if prelude && offset < child.offset {
 			continue
 		}
@@ -552,12 +562,6 @@ func (child *partitionConsumer) parseRecords(batch *RecordBatch) ([]*ConsumerMes
 	if incomplete {
 		return nil, ErrIncompleteResponse
 	}
-
-	child.offset = batch.FirstOffset + int64(batch.LastOffsetDelta) + 1
-	if child.offset <= originalOffset {
-		return nil, ErrConsumerOffsetNotAdvanced
-	}
-
 	return messages, nil
 }
 

record_batch.go

@@ -64,7 +64,7 @@ func (b *RecordBatch) encode(pe packetEncoder) error {
 	pe.putInt8(b.Version)
 	pe.push(newCRC32Field(crcCastagnoli))
 	pe.putInt16(b.computeAttributes())
-	pe.putInt32(int32(len(b.Records)))
+	pe.putInt32(b.LastOffsetDelta)
 
 	if err := (Timestamp{&b.FirstTimestamp}).encode(pe); err != nil {
 		return err

record_test.go

@@ -69,9 +69,10 @@ var recordBatchTestCases = []struct {
 	{
 		name: "uncompressed record",
 		batch: RecordBatch{
-			Version:        2,
-			FirstTimestamp: time.Unix(1479847795, 0),
-			MaxTimestamp:   time.Unix(0, 0),
+			Version:         2,
+			LastOffsetDelta: 1,
+			FirstTimestamp:  time.Unix(1479847795, 0),
+			MaxTimestamp:    time.Unix(0, 0),
 			Records: []*Record{{
 				TimestampDelta: 5 * time.Millisecond,
 				Key:            []byte{1, 2, 3, 4},
@@ -115,10 +116,11 @@ var recordBatchTestCases = []struct {
 	{
 		name: "gzipped record",
 		batch: RecordBatch{
-			Version:        2,
-			Codec:          CompressionGZIP,
-			FirstTimestamp: time.Unix(1479847795, 0),
-			MaxTimestamp:   time.Unix(0, 0),
+			Version:         2,
+			Codec:           CompressionGZIP,
+			LastOffsetDelta: 1,
+			FirstTimestamp:  time.Unix(1479847795, 0),
+			MaxTimestamp:    time.Unix(0, 0),
 			Records: []*Record{{
 				TimestampDelta: 5 * time.Millisecond,
 				Key:            []byte{1, 2, 3, 4},
@@ -168,10 +170,11 @@ var recordBatchTestCases = []struct {
 	{
 		name: "snappy compressed record",
 		batch: RecordBatch{
-			Version:        2,
-			Codec:          CompressionSnappy,
-			FirstTimestamp: time.Unix(1479847795, 0),
-			MaxTimestamp:   time.Unix(0, 0),
+			Version:         2,
+			Codec:           CompressionSnappy,
+			LastOffsetDelta: 1,
+			FirstTimestamp:  time.Unix(1479847795, 0),
+			MaxTimestamp:    time.Unix(0, 0),
 			Records: []*Record{{
 				TimestampDelta: 5 * time.Millisecond,
 				Key:            []byte{1, 2, 3, 4},
@@ -203,10 +206,11 @@ var recordBatchTestCases = []struct {
 	{
 		name: "lz4 compressed record",
 		batch: RecordBatch{
-			Version:        2,
-			Codec:          CompressionLZ4,
-			FirstTimestamp: time.Unix(1479847795, 0),
-			MaxTimestamp:   time.Unix(0, 0),
+			Version:         2,
+			Codec:           CompressionLZ4,
+			LastOffsetDelta: 1,
+			FirstTimestamp:  time.Unix(1479847795, 0),
+			MaxTimestamp:    time.Unix(0, 0),
 			Records: []*Record{{
 				TimestampDelta: 5 * time.Millisecond,
 				Key:            []byte{1, 2, 3, 4},