Internally, kafka-go offset is consistent with itself. Unfortunately, not with the rest of the world. The commit offset should be the offset of the next message to read and NOT the last message read. Verified by running sarama and kafka-go sequentially to verify they picked up each others offsets. (#62)

Seeing as you were working in commits, I thought I would hop on your commit.

Author: savaki

commit.go

@@ -14,7 +14,7 @@ func makeCommit(msg Message) commit {
 	return commit{
 		topic:     msg.Topic,
 		partition: msg.Partition,
-		offset:    msg.Offset,
+		offset:    msg.Offset + 1,
 	}
 }
 

commit_test.go

@@ -0,0 +1,22 @@
+package kafka
+
+import "testing"
+
+func TestMakeCommit(t *testing.T) {
+	msg := Message{
+		Topic:     "blah",
+		Partition: 1,
+		Offset:    2,
+	}
+
+	commit := makeCommit(msg)
+	if commit.topic != msg.Topic {
+		t.Errorf("bad topic: expected %v; got %v", msg.Topic, commit.topic)
+	}
+	if commit.partition != msg.Partition {
+		t.Errorf("bad partition: expected %v; got %v", msg.Partition, commit.partition)
+	}
+	if commit.offset != msg.Offset+1 {
+		t.Errorf("expected committed offset to be 1 greater than msg offset")
+	}
+}

reader.go

@@ -483,9 +483,6 @@ func (r *Reader) fetchOffsets(subs map[string][]int32) (map[int]int64, error) {
 		for _, partition := range partitions {
 			if partition == pr.Partition {
 				offset := pr.Offset
-				if offset >= 0 {
-					offset++ // advance to next offset
-				}
 				offsetsByPartition[int(partition)] = offset
 			}
 		}

reader_test.go

@@ -1081,16 +1081,16 @@ func TestOffsetStash(t *testing.T) {
 			Given:    offsetStash{},
 			Messages: []Message{newMessage(0, 0)},
 			Expected: offsetStash{
-				topic: {0: 0},
+				topic: {0: 1},
 			},
 		},
 		"ignores earlier offsets": {
 			Given: offsetStash{
-				topic: {0: 1},
+				topic: {0: 2},
 			},
 			Messages: []Message{newMessage(0, 0)},
 			Expected: offsetStash{
-				topic: {0: 1},
+				topic: {0: 2},
 			},
 		},
 		"uses latest offset": {
@@ -1101,7 +1101,7 @@ func TestOffsetStash(t *testing.T) {
 				newMessage(0, 1),
 			},
 			Expected: offsetStash{
-				topic: {0: 3},
+				topic: {0: 4},
 			},
 		},
 		"uses latest offset, across multiple topics": {
@@ -1115,8 +1115,8 @@ func TestOffsetStash(t *testing.T) {
 			},
 			Expected: offsetStash{
 				topic: {
-					0: 3,
-					1: 6,
+					0: 4,
+					1: 7,
 				},
 			},
 		},