Merge pull request #520 hide PublicErrCommitSessionToExpiredSession for async commit mode

Author: rekby

internal/topic/topicreaderinternal/stream_reader_impl.go

@@ -329,6 +329,12 @@ func (r *topicStreamReaderImpl) onPartitionSessionStatusResponseFromBuffer(
 }
 
 func (r *topicStreamReaderImpl) Commit(ctx context.Context, commitRange commitRange) (err error) {
+	defer func() {
+		if errors.Is(err, PublicErrCommitSessionToExpiredSession) && r.cfg.CommitMode == CommitModeAsync {
+			err = nil
+		}
+	}()
+
 	if commitRange.partitionSession == nil {
 		return xerrors.WithStackTrace(errCommitWithNilPartitionSession)
 	}
@@ -354,6 +360,9 @@ func (r *topicStreamReaderImpl) Commit(ctx context.Context, commitRange commitRa
 }
 
 func (r *topicStreamReaderImpl) checkCommitRange(commitRange commitRange) error {
+	if r.cfg.CommitMode == CommitModeNone {
+		return ErrCommitDisabled
+	}
 	session := commitRange.partitionSession
 
 	if session == nil {

internal/topic/topicreaderinternal/stream_reader_impl_test.go

@@ -679,29 +679,41 @@ func TestTopicStreamReadImpl_BatchReaderWantMoreMessagesThenBufferCanHold(t *tes
 }
 
 func TestTopicStreamReadImpl_CommitWithBadSession(t *testing.T) {
-	sleep := func() {
-		time.Sleep(time.Second / 10)
-	}
-	e := newTopicReaderTestEnv(t)
-	e.Start()
-
-	cr := commitRange{
-		partitionSession: newPartitionSession(
-			context.Background(),
-			"asd",
-			123,
-			nextReaderID(),
-			"bad-connection-id",
-			222,
-			213,
-		),
-	}
-	err := e.reader.Commit(e.ctx, cr)
-	require.Error(t, err)
+	commitByMode := func(mode PublicCommitMode) error {
+		sleep := func() {
+			time.Sleep(time.Second / 10)
+		}
+		e := newTopicReaderTestEnv(t)
+		e.reader.cfg.CommitMode = mode
+		e.Start()
 
-	sleep()
+		cr := commitRange{
+			partitionSession: newPartitionSession(
+				context.Background(),
+				"asd",
+				123,
+				nextReaderID(),
+				"bad-connection-id",
+				222,
+				213,
+			),
+		}
+		commitErr := e.reader.Commit(e.ctx, cr)
 
-	require.False(t, e.reader.closed)
+		sleep()
+
+		require.False(t, e.reader.closed)
+		return commitErr
+	}
+	t.Run("CommitModeNone", func(t *testing.T) {
+		require.ErrorIs(t, commitByMode(CommitModeNone), ErrCommitDisabled)
+	})
+	t.Run("CommitModeSync", func(t *testing.T) {
+		require.ErrorIs(t, commitByMode(CommitModeSync), PublicErrCommitSessionToExpiredSession)
+	})
+	t.Run("CommitModeAsync", func(t *testing.T) {
+		require.NoError(t, commitByMode(CommitModeAsync))
+	})
 }
 
 type streamEnv struct {