add CommitMessages documentation (segmentio#58)

* add CommitMessages documentation

* Revisit sync commits (segmentio#59)

* revisit sync commits

* drop commit request chan capacity

* remove error prone check

* shorter code

Author: achille-roussel

commit.go

@@ -0,0 +1,39 @@
+package kafka
+
+// A commit represents the instruction of publishing an update of the last
+// offset read by a program for a topic and partition.
+type commit struct {
+	topic     string
+	partition int
+	offset    int64
+}
+
+// makeCommit builds a commit value from a message, the resulting commit takes
+// its topic, partition, and offset from the message.
+func makeCommit(msg Message) commit {
+	return commit{
+		topic:     msg.Topic,
+		partition: msg.Partition,
+		offset:    msg.Offset,
+	}
+}
+
+// makeCommits generates a slice of commits from a list of messages, it extracts
+// the topic, partition, and offset of each message and builds the corresponding
+// commit slice.
+func makeCommits(msgs ...Message) []commit {
+	commits := make([]commit, len(msgs))
+
+	for i, m := range msgs {
+		commits[i] = makeCommit(m)
+	}
+
+	return commits
+}
+
+// commitRequest is the data type exchanged between the CommitMessages method
+// and internals of the reader's implementation.
+type commitRequest struct {
+	commits []commit
+	errch   chan<- error
+}

reader.go

@@ -77,7 +77,7 @@ type Reader struct {
 	cancel       context.CancelFunc
 	stop         context.CancelFunc
 	done         chan struct{}
-	commits      chan []Message
+	commits      chan commitRequest
 	version      int64 // version holds the generation of the spawned readers
 	offset       int64
 	lag          int64
@@ -96,10 +96,12 @@ type Reader struct {
 	stats readerStats
 }
 
-// useConsumerGroup indicates the Reader is part of a consumer group
-func (r *Reader) useConsumerGroup() bool {
-	return r.config.GroupID != ""
-}
+// useConsumerGroup indicates whether the Reader is part of a consumer group.
+func (r *Reader) useConsumerGroup() bool { return r.config.GroupID != "" }
+
+// useSyncCommits indicates whether the Reader is configured to perform sync or
+// async commits.
+func (r *Reader) useSyncCommits() bool { return r.config.CommitInterval == 0 }
 
 // membership returns the group generationID and memberID of the reader.
 //
@@ -676,20 +678,16 @@ func (r *Reader) commitOffsetsWithRetry(conn offsetCommitter, offsetStash offset
 type offsetStash map[string]map[int]int64
 
 // merge updates the offsetStash with the offsets from the provided messages
-func (o offsetStash) merge(msgs ...Message) {
-	if o == nil {
-		return
-	}
-
-	for _, m := range msgs {
-		offsetsByPartition, ok := o[m.Topic]
+func (o offsetStash) merge(commits []commit) {
+	for _, c := range commits {
+		offsetsByPartition, ok := o[c.topic]
 		if !ok {
 			offsetsByPartition = map[int]int64{}
-			o[m.Topic] = offsetsByPartition
+			o[c.topic] = offsetsByPartition
 		}
 
-		if offset, ok := offsetsByPartition[m.Partition]; !ok || m.Offset > offset {
-			offsetsByPartition[m.Partition] = m.Offset
+		if offset, ok := offsetsByPartition[c.partition]; !ok || c.offset > offset {
+			offsetsByPartition[c.partition] = c.offset
 		}
 	}
 }
@@ -701,35 +699,19 @@ func (o offsetStash) reset() {
 	}
 }
 
-// isEmpty returns true if the offsetStash contains no entries
-func (o offsetStash) isEmpty() bool {
-	return len(o) == 0
-}
-
 // commitLoopImmediate handles each commit synchronously
 func (r *Reader) commitLoopImmediate(conn offsetCommitter, stop <-chan struct{}) {
+	offsetsByTopicAndPartition := offsetStash{}
+
 	for {
 		select {
 		case <-stop:
 			return
 
-		case msgs, ok := <-r.commits:
-			if !ok {
-				r.withErrorLogger(func(l *log.Logger) {
-					l.Println("reader commit channel unexpectedly closed")
-				})
-				return
-			}
-
-			offsetsByTopicAndPartition := offsetStash{}
-			offsetsByTopicAndPartition.merge(msgs...)
-
-			if err := r.commitOffsetsWithRetry(conn, offsetsByTopicAndPartition, defaultCommitRetries); err != nil {
-				r.withErrorLogger(func(l *log.Logger) {
-					l.Printf("unable to commit offset: %v", err)
-				})
-				return
-			}
+		case req := <-r.commits:
+			offsetsByTopicAndPartition.merge(req.commits)
+			req.errch <- r.commitOffsetsWithRetry(conn, offsetsByTopicAndPartition, defaultCommitRetries)
+			offsetsByTopicAndPartition.reset()
 		}
 	}
 }
@@ -740,40 +722,25 @@ func (r *Reader) commitLoopInterval(conn offsetCommitter, stop <-chan struct{})
 	ticker := time.NewTicker(r.config.HeartbeatInterval)
 	defer ticker.Stop()
 
-	defer func() {
-		// commits any outstanding offsets on close
-		if err := r.commitOffsetsWithRetry(conn, r.offsetStash, defaultCommitRetries); err == nil {
+	commit := func() {
+		if err := r.commitOffsetsWithRetry(conn, r.offsetStash, defaultCommitRetries); err != nil {
+			r.withErrorLogger(func(l *log.Logger) { l.Print(err) })
+		} else {
 			r.offsetStash.reset()
 		}
-	}()
+	}
 
 	for {
 		select {
 		case <-stop:
+			commit()
 			return
 
 		case <-ticker.C:
-			if len(r.offsetStash) == 0 {
-				continue
-			}
-
-			if err := r.commitOffsetsWithRetry(conn, r.offsetStash, defaultCommitRetries); err != nil {
-				r.withErrorLogger(func(l *log.Logger) {
-					l.Printf("unable to commit offset: %v", err)
-				})
-				return
-			}
-			r.offsetStash.reset()
-
-		case msgs, ok := <-r.commits:
-			if !ok {
-				r.withErrorLogger(func(l *log.Logger) {
-					l.Println("reader commit channel unexpectedly closed")
-				})
-				return
-			}
+			commit()
 
-			r.offsetStash.merge(msgs...)
+		case req := <-r.commits:
+			r.offsetStash.merge(req.commits)
 		}
 	}
 }
@@ -1106,7 +1073,7 @@ func NewReader(config ReaderConfig) *Reader {
 		msgs:    make(chan readerMessage, config.QueueCapacity),
 		cancel:  func() {},
 		done:    make(chan struct{}),
-		commits: make(chan []Message),
+		commits: make(chan commitRequest),
 		stop:    stop,
 		offset:  firstOffset,
 		stctx:   stctx,
@@ -1247,6 +1214,47 @@ func (r *Reader) FetchMessage(ctx context.Context) (Message, error) {
 	}
 }
 
+// CommitMessages commits the list of messages passed as argument. The program
+// may pass a context to asynchronously cancel the commit operation when it was
+// configured to be blocking.
+func (r *Reader) CommitMessages(ctx context.Context, msgs ...Message) error {
+	if !r.useConsumerGroup() {
+		return errNotAvailable
+	}
+
+	var errch <-chan error
+	var sync = r.useSyncCommits()
+	var creq = commitRequest{
+		commits: makeCommits(msgs...),
+	}
+
+	if sync {
+		ch := make(chan error, 1)
+		errch, creq.errch = ch, ch
+	}
+
+	select {
+	case r.commits <- creq:
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-r.stctx.Done():
+		// This context is used to ensure we don't allow commits after the
+		// reader was closed.
+		return io.ErrClosedPipe
+	}
+
+	if !sync {
+		return nil
+	}
+
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case err := <-errch:
+		return err
+	}
+}
+
 // ReadLag returns the current lag of the reader by fetching the last offset of
 // the topic and partition and computing the difference between that value and
 // the offset of the last message returned by ReadMessage.
@@ -1501,25 +1509,6 @@ func (r *Reader) start(offsetsByPartition map[int]int64) {
 	}
 }
 
-func (r *Reader) CommitMessages(ctx context.Context, msgs ...Message) error {
-	if len(msgs) == 0 {
-		return nil
-	}
-
-	if !r.useConsumerGroup() {
-		return errNotAvailable
-	}
-
-	select {
-	case <-ctx.Done():
-		return ctx.Err()
-	case <-r.stctx.Done():
-		return r.stctx.Err()
-	case r.commits <- msgs:
-		return nil
-	}
-}
-
 // A reader reads messages from kafka and produces them on its channels, it's
 // used as an way to asynchronously fetch messages while the main program reads
 // them using the high level reader API.

reader_test.go

@@ -1124,7 +1124,7 @@ func TestOffsetStash(t *testing.T) {
 
 	for label, test := range tests {
 		t.Run(label, func(t *testing.T) {
-			test.Given.merge(test.Messages...)
+			test.Given.merge(makeCommits(test.Messages...))
 			if !reflect.DeepEqual(test.Expected, test.Given) {
 				t.Errorf("expected %v; got %v", test.Expected, test.Given)
 			}