Merge pull request #1837 from ydb-platform/1802-bad-offsets

Fixed respect start offset from topicoptions.WithReaderGetPartitionStartOffset for commit messages

Author: rekby

CHANGELOG.md

@@ -1,3 +1,4 @@
+* Fixed respect start offset from topicoptions.WithReaderGetPartitionStartOffset for commit messages
 * Added `query.AllowImplicitSessions()` option for execute queries through `query.Client.{Exec,Query,QueryResultSet,QueryRow}` without explicit sessions
 
 ## v3.112.0

internal/topic/topicreadercommon/read_partition_session.go

@@ -48,8 +48,7 @@ func NewPartitionSession(
 		StreamPartitionSessionID: partitionSessionID,
 		ClientPartitionSessionID: clientPartitionSessionID,
 	}
-	res.committedOffsetVal.Store(committedOffset.ToInt64())
-	res.lastReceivedOffsetEndVal.Store(committedOffset.ToInt64() - 1)
+	res.SetInitialCommitOffset(committedOffset)
 
 	return res
 }
@@ -103,6 +102,11 @@ func (s *PartitionSession) SetLastReceivedMessageOffset(v rawtopiccommon.Offset)
 	s.lastReceivedOffsetEndVal.Store(v.ToInt64())
 }
 
+func (s *PartitionSession) SetInitialCommitOffset(committedOffset rawtopiccommon.Offset) {
+	s.committedOffsetVal.Store(committedOffset.ToInt64())
+	s.lastReceivedOffsetEndVal.Store(committedOffset.ToInt64() - 1)
+}
+
 func (s *PartitionSession) NoMoreMessages() bool {
 	return s.noMoreMessages.Load()
 }

internal/topic/topicreaderinternal/stream_reader_impl.go

@@ -991,6 +991,9 @@ func (r *topicStreamReaderImpl) onStartPartitionSessionRequestFromBuffer(
 		commitOffset = forceOffset
 		respMessage.CommitOffset.FromInt64Pointer(commitOffset)
 	}
+	if forceOffset != nil {
+		session.SetInitialCommitOffset(rawtopiccommon.NewOffset(*forceOffset))
+	}
 
 	return r.send(respMessage)
 }

tests/integration/topic_read_messages_test.go

@@ -4,12 +4,17 @@
 package integration
 
 import (
+	"context"
 	"testing"
+	"time"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xtest"
+	"github.com/ydb-platform/ydb-go-sdk/v3/topic/topicoptions"
 	"github.com/ydb-platform/ydb-go-sdk/v3/topic/topicsugar"
+	"github.com/ydb-platform/ydb-go-sdk/v3/topic/topicwriter"
 )
 
 func TestTopicReadMessages(t *testing.T) {
@@ -35,3 +40,45 @@ func TestTopicReadMessages(t *testing.T) {
 	require.NoError(t, err)
 	require.NotEmpty(t, batch.Messages)
 }
+
+func TestRegression1802_StartPartitionWithOffsetHandler(t *testing.T) {
+	scope := newScope(t)
+	ctx := scope.Ctx
+
+	scope.Require.NoError(scope.TopicWriter().Write(ctx,
+		topicwriter.Message{},
+		topicwriter.Message{},
+		topicwriter.Message{},
+	))
+
+	msg, err := scope.TopicReaderNamed("first").ReadMessage(ctx)
+	scope.Require.NoError(err)
+	err = scope.TopicReaderNamed("first").Commit(ctx, msg)
+	scope.Require.NoError(err)
+	err = scope.TopicReaderNamed("first").Close(ctx)
+	scope.Require.NoError(err)
+
+	reader, err := scope.DriverWithGRPCLogging().Topic().StartReader(
+		scope.TopicConsumerName(), topicoptions.ReadTopic(scope.TopicPath()),
+		topicoptions.WithReaderGetPartitionStartOffset(func(ctx context.Context, req topicoptions.GetPartitionStartOffsetRequest) (res topicoptions.GetPartitionStartOffsetResponse, err error) {
+			res.StartFrom(2)
+			return res, nil
+		}),
+	)
+	scope.Require.NoError(err)
+
+	msg, err = reader.ReadMessage(ctx)
+	scope.Require.NoError(err)
+
+	scope.Logf("Received message offset: %v", msg.Offset)
+
+	err = reader.Commit(ctx, msg)
+	scope.Require.NoError(err)
+
+	scope.Require.EventuallyWithT(func(t *assert.CollectT) {
+		c, err := scope.Driver().Topic().DescribeTopicConsumer(ctx, scope.TopicPath(), scope.TopicConsumerName(), topicoptions.IncludeConsumerStats())
+		require.NoError(t, err)
+
+		require.EqualValues(t, 3, c.Partitions[0].PartitionConsumerStats.CommittedOffset)
+	}, 10*time.Second, 100*time.Millisecond)
+}