Fix test errors.

Author: nachogiljaldo

consumergroup.go

@@ -931,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 {
@@ -1079,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)

error.go

@@ -34,7 +34,7 @@ const (
 	NotEnoughReplicas                  Error = 19
 	NotEnoughReplicasAfterAppend       Error = 20
 	InvalidRequiredAcks                Error = 21
-	IllegalGenerationErr               Error = 22
+	IllegalGeneration                  Error = 22
 	InconsistentGroupProtocol          Error = 23
 	InvalidGroupId                     Error = 24
 	UnknownMemberId                    Error = 25
@@ -217,7 +217,7 @@ func (e Error) Title() string {
 		return "Not Enough Replicas After Append"
 	case InvalidRequiredAcks:
 		return "Invalid Required Acks"
-	case IllegalGenerationErr:
+	case IllegalGeneration:
 		return "Illegal Generation"
 	case InconsistentGroupProtocol:
 		return "Inconsistent Group Protocol"
@@ -426,7 +426,7 @@ func (e Error) Description() string {
 		return "the message was written to the log, but with fewer in-sync replicas than required."
 	case InvalidRequiredAcks:
 		return "the requested requiredAcks is invalid (anything other than -1, 1, or 0)"
-	case IllegalGenerationErr:
+	case IllegalGeneration:
 		return "the generation id provided in the request is not the current generation"
 	case InconsistentGroupProtocol:
 		return "the member provided a protocol type or set of protocols which is not compatible with the current group"

error_test.go

@@ -28,7 +28,7 @@ func TestError(t *testing.T) {
 		NotEnoughReplicas,
 		NotEnoughReplicasAfterAppend,
 		InvalidRequiredAcks,
-		IllegalGenerationErr,
+		IllegalGeneration,
 		InconsistentGroupProtocol,
 		InvalidGroupId,
 		UnknownMemberId,

reader.go

@@ -174,23 +174,24 @@ func (r *Reader) commitOffsetsWithRetry(gen *Generation, offsetStash offsetStash
 			}
 
 			if err = gen.CommitOffsetsForGenID(generationID, messages); err == nil {
-				continue
+				break
 			}
 
 			// IllegalGeneration error is not retriable, but we should attempt to
 			// perform the remaining commits
-			if errors.Is(err, IllegalGenerationErr) {
+			if errors.Is(err, IllegalGeneration) {
 				r.withErrorLogger(func(l Logger) { l.Printf("%v", err) })
 				illegalGenerationErr = true
 				err = nil
 				offsetStash.removeGenerationID(generationID)
 			}
+			return
 		}
 	}
 
 	// if configured to ignore the error
 	if illegalGenerationErr && r.config.ErrorOnWrongGenerationCommit {
-		err = IllegalGenerationErr
+		err = IllegalGeneration
 	}
 	return // err will not be nil
 }

reader_test.go

@@ -1261,7 +1261,7 @@ func testReaderConsumerGroupRebalanceDoesNotCommitNotOwnedPartitions(t *testing.
 	require.NoError(t, secondReader.CommitMessages(ctx, msgsForSecondReader...))
 
 	// commit all messages the first reader received, we expect an error
-	require.ErrorIs(t, IllegalGenerationErr, firstReader.CommitMessages(ctx, allMessages...))
+	require.ErrorIs(t, IllegalGeneration, firstReader.CommitMessages(ctx, allMessages...))
 
 	// verify that no offsets have been altered
 	require.Eventually(t, func() bool {