Fail attempts to commit offsets when the generationID is not the same.

Author: nachogiljaldo

consumergroup.go

@@ -416,6 +416,12 @@ func (g *Generation) Start(fn func(ctx context.Context)) {
 // consumer group coordinator.  This can be used to reset the consumer to
 // explicit offsets.
 func (g *Generation) CommitOffsets(offsets map[string]map[int]int64) error {
+	return g.CommitOffsetsForGenID(g.ID, offsets)
+}
+
+// CommitOffsetsForGenID commits the provided topic+partition+offset combos to the
+// consumer group coordinator specifying the given genID.
+func (g *Generation) CommitOffsetsForGenID(genID int32, offsets map[string]map[int]int64) error {
 	if len(offsets) == 0 {
 		return nil
 	}
@@ -434,7 +440,7 @@ func (g *Generation) CommitOffsets(offsets map[string]map[int]int64) error {
 
 	request := offsetCommitRequestV2{
 		GroupID:       g.GroupID,
-		GenerationID:  g.ID,
+		GenerationID:  genID,
 		MemberID:      g.MemberID,
 		RetentionTime: g.retentionMillis,
 		Topics:        topics,
@@ -925,12 +931,12 @@ func (cg *ConsumerGroup) coordinator() (coordinator, error) {
 // the leader.  Otherwise, GroupMemberAssignments will be nil.
 //
 // Possible kafka error codes returned:
-//  * GroupLoadInProgress:
-//  * GroupCoordinatorNotAvailable:
-//  * NotCoordinatorForGroup:
-//  * InconsistentGroupProtocol:
-//  * InvalidSessionTimeout:
-//  * GroupAuthorizationFailed:
+//   - GroupLoadInProgress:
+//   - GroupCoordinatorNotAvailable:
+//   - NotCoordinatorForGroup:
+//   - InconsistentGroupProtocol:
+//   - InvalidSessionTimeout:
+//   - GroupAuthorizationFailed:
 func (cg *ConsumerGroup) joinGroup(conn coordinator, memberID string) (string, int32, GroupMemberAssignments, error) {
 	request, err := cg.makeJoinGroupRequestV1(memberID)
 	if err != nil {
@@ -1073,11 +1079,11 @@ func (cg *ConsumerGroup) makeMemberProtocolMetadata(in []joinGroupResponseMember
 // Readers subscriptions topic => partitions
 //
 // Possible kafka error codes returned:
-//  * GroupCoordinatorNotAvailable:
-//  * NotCoordinatorForGroup:
-//  * IllegalGeneration:
-//  * RebalanceInProgress:
-//  * GroupAuthorizationFailed:
+//   - GroupCoordinatorNotAvailable:
+//   - NotCoordinatorForGroup:
+//   - IllegalGeneration:
+//   - RebalanceInProgress:
+//   - GroupAuthorizationFailed:
 func (cg *ConsumerGroup) syncGroup(conn coordinator, memberID string, generationID int32, memberAssignments GroupMemberAssignments) (map[string][]int32, error) {
 	request := cg.makeSyncGroupRequestV0(memberID, generationID, memberAssignments)
 	response, err := conn.syncGroup(request)

reader.go

@@ -150,54 +150,53 @@ func (r *Reader) commitOffsetsWithRetry(gen *Generation, offsetStash offsetStash
 		backoffDelayMax = 5 * time.Second
 	)
 
-	messagesToSend := make(map[string]map[int]int64)
+	messagesToSendForGeneration := make(map[int32]map[string]map[int]int64)
 	for topic, partitionsInfo := range offsetStash {
-		msgsForTopic := make(map[int]int64)
 		for partition, commitInfo := range partitionsInfo {
-			// if there is a generation and it is different than the current
-			// it means there was a rebalance
-			if commitInfo.generationId != undefinedGenerationId {
-				if gen.ID == commitInfo.generationId {
-					msgsForTopic[partition] = commitInfo.offset
-				} else {
-					if assignments, ok := gen.Assignments[topic]; ok {
-						for _, assignment := range assignments {
-							if assignment.ID == partition && commitInfo.generationId > gen.ID {
-								msgsForTopic[partition] = commitInfo.offset
-							}
-						}
-					}
-					r.withErrorLogger(func(l Logger) {
-						l.Printf("Discarding commint for %s - %d: %d . Current generation is %d, commit generation is %d", topic, partition, commitInfo.offset, gen.ID, commitInfo.generationId)
-					})
-				}
-			} else {
-				msgsForTopic[partition] = commitInfo.offset
+			if _, ok := messagesToSendForGeneration[commitInfo.generationID]; !ok {
+				messagesToSendForGeneration[commitInfo.generationID] = make(map[string]map[int]int64)
 			}
-		}
-		if len(msgsForTopic) > 0 {
-			messagesToSend[topic] = msgsForTopic
+			msgsForTopic := messagesToSendForGeneration[commitInfo.generationID]
+			if _, ok := msgsForTopic[topic]; !ok {
+				msgsForTopic[topic] = make(map[int]int64)
+			}
+			msgsForPartition := msgsForTopic[topic]
+			msgsForPartition[partition] = commitInfo.offset
 		}
 	}
-	for attempt := 0; attempt < retries; attempt++ {
-		if attempt != 0 {
-			if !sleep(r.stctx, backoff(attempt, backoffDelayMin, backoffDelayMax)) {
-				return
+	var illegalGenerationErr error
+	for generationID, messages := range messagesToSendForGeneration {
+		for attempt := 0; attempt < retries; attempt++ {
+			if attempt != 0 {
+				if !sleep(r.stctx, backoff(attempt, backoffDelayMin, backoffDelayMax)) {
+					continue
+				}
 			}
-		}
 
-		if err = gen.CommitOffsets(messagesToSend); err == nil {
-			return
+			if err = gen.CommitOffsetsForGenID(generationID, messages); err == nil {
+				continue
+			}
+			// IllegalGeneration error is not retriable, but we should attempt to
+			// perform the remaining commits
+			if err == IllegalGeneration {
+				illegalGenerationErr = err
+				// we prevent useless retries and we will attempt to
+				// commit the remaining generations.
+				err = nil
+				offsetStash.removeGenerationID(generationID)
+			}
 		}
 	}
-
+	if illegalGenerationErr != nil {
+		err = illegalGenerationErr
+	}
 	return // err will not be nil
 }
 
 // offsetStash holds offsets by topic => partition => offsetEntry.
 type offsetEntry struct {
 	offset       int64
-	generationId int32
+	generationID int32
 }
 type offsetStash map[string]map[int]offsetEntry
 
@@ -213,7 +212,7 @@ func (o offsetStash) merge(commits []commit) {
 		if offset, ok := offsetsByPartition[c.partition]; !ok || c.offset > offset.offset {
 			offsetsByPartition[c.partition] = offsetEntry{
 				offset:       c.offset,
-				generationId: c.generationId,
+				generationID: c.generationId,
 			}
 		}
 	}
@@ -226,6 +225,16 @@ func (o offsetStash) reset() {
 	}
 }
 
+func (o offsetStash) removeGenerationID(genID int32) {
+	for _, offsetsForTopic := range o {
+		for partition, offsetsForPartition := range offsetsForTopic {
+			if offsetsForPartition.generationID == genID {
+				delete(offsetsForTopic, partition)
+			}
+		}
+	}
+}
+
 // commitLoopImmediate handles each commit synchronously.
 func (r *Reader) commitLoopImmediate(ctx context.Context, gen *Generation) {
 	offsets := offsetStash{}

reader_test.go

@@ -856,11 +856,11 @@ func TestReaderConsumerGroup(t *testing.T) {
 			partitions: 1,
 			function:   testConsumerGroupSimple,
 		},
-
 		{
-			scenario:   "Do not commit not assigned messages after rebalance",
-			partitions: 2,
-			function:   testReaderConsumerGroupRebalanceDoesNotCommitNotOwnedPartitions,
+			scenario:       "Do not commit not assigned messages after rebalance",
+			partitions:     2,
+			function:       testReaderConsumerGroupRebalanceDoesNotCommitNotOwnedPartitions,
+			commitInterval: 0,
 		},
 	}
 
@@ -1203,6 +1203,7 @@ func testReaderConsumerGroupRebalanceDoesNotCommitNotOwnedPartitions(t *testing.
 
 	// read all messages for the first reader
 	msgsForFirstReader := make(map[int][]Message, 0)
+	allMessages := make([]Message, 0)
 	totalEvents := 0
 	for i := 0; i < messageCount; i++ {
 		if msg, err := firstReader.FetchMessage(ctx); err != nil {
@@ -1213,6 +1214,7 @@ func testReaderConsumerGroupRebalanceDoesNotCommitNotOwnedPartitions(t *testing.
 				msgs = make([]Message, 0)
 			}
 			msgs = append(msgs, msg)
+			allMessages = append(allMessages, msg)
 			msgsForFirstReader[msg.Partition] = msgs
 			totalEvents++
 		}
@@ -1265,8 +1267,8 @@ func testReaderConsumerGroupRebalanceDoesNotCommitNotOwnedPartitions(t *testing.
 		return false
 	}, 5*time.Second, 100*time.Millisecond, "Offsets were never committed")
 
-	// commit first message for the second reader on the first reader
-	require.NoError(t, firstReader.CommitMessages(ctx, msgsForFirstReader[partitionAssignedToSecondConsumer][0]))
+	// commit all messages the first reader received
+	require.ErrorIs(t, IllegalGeneration, firstReader.CommitMessages(ctx, allMessages...))
 	require.Eventually(t, func() bool {
 		resp, err := client.OffsetFetch(ctx, &OffsetFetchRequest{
 			GroupID: firstReader.config.GroupID,