Make erroring optional and simplify the test.

Author: nachogiljaldo

reader.go

@@ -179,6 +179,7 @@ func (r *Reader) commitOffsetsWithRetry(gen *Generation, offsetStash offsetStash
 			// IllegalGeneration error is not retriable, but we should attempt to
 			// perform the remaining commits
 			if err == IllegalGeneration {
+				r.withErrorLogger(func(l Logger) { l.Printf("%v", err) })
 				illegalGenerationErr = err
 				// we prevent useless retries and we will attempt to
 				// commit the remaining generations.
@@ -187,7 +188,9 @@ func (r *Reader) commitOffsetsWithRetry(gen *Generation, offsetStash offsetStash
 			}
 		}
 	}
-	if illegalGenerationErr != nil {
+
+	// if configured to ignore the error
+	if illegalGenerationErr != nil && r.config.ErrorOnWrongGenerationCommit {
 		err = illegalGenerationErr
 	}
 	return // err will not be nil
@@ -567,6 +570,10 @@ type ReaderConfig struct {
 	// This flag is being added to retain backwards-compatibility, so it will be
 	// removed in a future version of kafka-go.
 	OffsetOutOfRangeError bool
+
+	// ErrorOnWrongGenerationCommit indicates that we should return an error when
+	// attempting to commit a message to a generation different than the current one.
+	ErrorOnWrongGenerationCommit bool
 }
 
 // Validate method validates ReaderConfig properties.

reader_test.go

@@ -792,10 +792,11 @@ func TestExtractTopics(t *testing.T) {
 
 func TestReaderConsumerGroup(t *testing.T) {
 	tests := []struct {
-		scenario       string
-		partitions     int
-		commitInterval time.Duration
-		function       func(*testing.T, context.Context, *Reader)
+		scenario               string
+		partitions             int
+		commitInterval         time.Duration
+		errorOnWrongGeneration bool
+		function               func(*testing.T, context.Context, *Reader)
 	}{
 		{
 			scenario:   "basic handshake",
@@ -856,11 +857,13 @@ func TestReaderConsumerGroup(t *testing.T) {
 			partitions: 1,
 			function:   testConsumerGroupSimple,
 		},
+
 		{
-			scenario:       "Do not commit not assigned messages after rebalance",
-			partitions:     2,
-			function:       testReaderConsumerGroupRebalanceDoesNotCommitNotOwnedPartitions,
-			commitInterval: 0,
+			scenario:               "Do not commit not assigned messages after rebalance",
+			partitions:             2,
+			function:               testReaderConsumerGroupRebalanceDoesNotCommitNotOwnedPartitions,
+			commitInterval:         0,
+			errorOnWrongGeneration: true,
 		},
 	}
 
@@ -877,15 +880,16 @@ func TestReaderConsumerGroup(t *testing.T) {
 
 			groupID := makeGroupID()
 			r := NewReader(ReaderConfig{
-				Brokers:           []string{"localhost:9092"},
-				Topic:             topic,
-				GroupID:           groupID,
-				HeartbeatInterval: 2 * time.Second,
-				CommitInterval:    test.commitInterval,
-				RebalanceTimeout:  2 * time.Second,
-				RetentionTime:     time.Hour,
-				MinBytes:          1,
-				MaxBytes:          1e6,
+				Brokers:                      []string{"localhost:9092"},
+				Topic:                        topic,
+				GroupID:                      groupID,
+				HeartbeatInterval:            2 * time.Second,
+				CommitInterval:               test.commitInterval,
+				RebalanceTimeout:             2 * time.Second,
+				RetentionTime:                time.Hour,
+				MinBytes:                     1,
+				MaxBytes:                     1e6,
+				ErrorOnWrongGenerationCommit: test.errorOnWrongGeneration,
 			})
 			defer r.Close()
 
@@ -1193,6 +1197,8 @@ func testReaderConsumerGroupRebalanceDoesNotCommitNotOwnedPartitions(t *testing.
 		BatchSize: 1,
 		Transport: client.Transport,
 	}
+
+	// Write 4 messages and ensure that they go the each one of the partitions
 	messageCount := 4
 	if err := writer.WriteMessages(ctx, makeTestSequence(messageCount)...); err != nil {
 		t.Fatalf("bad write messages: %v", err)
@@ -1220,9 +1226,12 @@ func testReaderConsumerGroupRebalanceDoesNotCommitNotOwnedPartitions(t *testing.
 		}
 	}
 	require.Equal(t, messageCount, totalEvents)
+
+	// create a second reader
 	secondReader := NewReader(firstReader.config)
 	defer secondReader.Close()
 
+	// wait until the group has 2 members
 	require.Eventually(t, func() bool {
 		resp, err := client.DescribeGroups(
 			ctx,
@@ -1247,28 +1256,14 @@ func testReaderConsumerGroupRebalanceDoesNotCommitNotOwnedPartitions(t *testing.
 	}
 	partitionAssignedToFirstConsumer := (partitionAssignedToSecondConsumer + 1) % 2
 
-	// commit all messages for the second reader and wait until commits reach the server
+	// commit all messages for the second reader (no need to wait until commits reach the server
+	// because CommitInterval is set to 0)
 	require.NoError(t, secondReader.CommitMessages(ctx, msgsForSecondReader...))
-	require.Eventually(t, func() bool {
-		resp, err := client.OffsetFetch(ctx, &OffsetFetchRequest{
-			GroupID: firstReader.config.GroupID,
-			Topics:  map[string][]int{firstReader.config.Topic: {partitionAssignedToSecondConsumer}},
-		})
-		require.NoError(t, err)
-		require.NotNil(t, resp)
 
-		for _, topicOffsets := range resp.Topics {
-			for _, offsetPartition := range topicOffsets {
-				if offsetPartition.Partition == partitionAssignedToSecondConsumer {
-					return msgsForSecondReader[len(msgsForSecondReader)-1].Offset+1 == offsetPartition.CommittedOffset
-				}
-			}
-		}
-		return false
-	}, 5*time.Second, 100*time.Millisecond, "Offsets were never committed")
-
-	// commit all messages the first reader received
+	// commit all messages the first reader received, we expect an error
 	require.ErrorIs(t, IllegalGeneration, firstReader.CommitMessages(ctx, allMessages...))
+
+	// verify that no offsets have been altered
 	require.Eventually(t, func() bool {
 		resp, err := client.OffsetFetch(ctx, &OffsetFetchRequest{
 			GroupID: firstReader.config.GroupID,
@@ -1287,7 +1282,9 @@ func testReaderConsumerGroupRebalanceDoesNotCommitNotOwnedPartitions(t *testing.
 		return false
 	}, 5*time.Second, 100*time.Millisecond, "Offsets were altered")
 
-	// commit the messages it can actually commit and verify it works
+	// we can read the messages again because generation changes
+	// cause uncommitted offsets to be lost
+	totalEvents = 0
 	for i := 0; i < len(msgsForFirstReader); i++ {
 		if msg, err := firstReader.FetchMessage(ctx); err != nil {
 			t.Errorf("reader %v expected to read 1 message", i)
@@ -1301,24 +1298,11 @@ func testReaderConsumerGroupRebalanceDoesNotCommitNotOwnedPartitions(t *testing.
 			totalEvents++
 		}
 	}
-	require.NoError(t, firstReader.CommitMessages(ctx, msgsForFirstReader[partitionAssignedToFirstConsumer][len(msgsForFirstReader[partitionAssignedToFirstConsumer])-1]))
-	require.Eventually(t, func() bool {
-		resp, err := client.OffsetFetch(ctx, &OffsetFetchRequest{
-			GroupID: firstReader.config.GroupID,
-			Topics:  map[string][]int{firstReader.config.Topic: {partitionAssignedToFirstConsumer}},
-		})
-		require.NoError(t, err)
-		require.NotNil(t, resp)
+	require.Equal(t, 2, totalEvents)
 
-		for _, topicOffsets := range resp.Topics {
-			for _, offsetPartition := range topicOffsets {
-				if offsetPartition.Partition == partitionAssignedToFirstConsumer {
-					return msgsForFirstReader[partitionAssignedToFirstConsumer][len(msgsForSecondReader)-1].Offset+1 == offsetPartition.CommittedOffset
-				}
-			}
-		}
-		return false
-	}, 5*time.Second, 100*time.Millisecond, "Could not commit the new element")
+	// commit the messages it can actually commit and verify it works
+	// no need to wait because CommitInterval is 0
+	require.NoError(t, firstReader.CommitMessages(ctx, msgsForFirstReader[partitionAssignedToFirstConsumer][len(msgsForFirstReader[partitionAssignedToFirstConsumer])-1]))
 }
 
 func TestOffsetStash(t *testing.T) {