Merge pull request #1018 Fixed init info in topic writer, when autoseq num turned off.

Author: rekby

CHANGELOG.md

@@ -1,3 +1,5 @@
+* Fixed init info in topic writer, when autoseq num turned off.
+
 ## v3.55.1
 * Supported column name prefix `__discard_column_` for discard columns in result sets
 * Made `StatusIds_SESSION_EXPIRED` retriable for idempotent operations

internal/topic/topicwriterinternal/writer_reconnector.go

@@ -428,7 +428,7 @@ func (w *WriterReconnector) startWriteStream(ctx, streamCtx context.Context, att
 }
 
 func (w *WriterReconnector) needReceiveLastSeqNo() bool {
-	res := w.cfg.AutoSetSeqNo && !w.firstConnectionHandled.Load()
+	res := !w.firstConnectionHandled.Load()
 	return res
 }
 
@@ -490,7 +490,7 @@ func (w *WriterReconnector) onWriterChange(writerStream *SingleStreamWriter) {
 		defer close(w.firstInitResponseProcessedChan)
 		isFirstInit = true
 
-		if w.cfg.AutoSetSeqNo {
+		if writerStream.LastSeqNumRequested {
 			w.lastSeqNo = writerStream.ReceivedLastSeqNum
 		}
 	})

internal/topic/topicwriterinternal/writer_reconnector_test.go

@@ -328,8 +328,9 @@ func TestWriterImpl_InitSession(t *testing.T) {
 	sessionID := "test-session-id"
 
 	w.onWriterChange(&SingleStreamWriter{
-		ReceivedLastSeqNum: lastSeqNo,
-		SessionID:          sessionID,
+		ReceivedLastSeqNum:  lastSeqNo,
+		LastSeqNumRequested: true,
+		SessionID:           sessionID,
 	})
 
 	require.Equal(t, sessionID, w.sessionID)
@@ -344,7 +345,8 @@ func TestWriterImpl_WaitInit(t *testing.T) {
 			LastSeqNum: int64(123),
 		}
 		w.onWriterChange(&SingleStreamWriter{
-			ReceivedLastSeqNum: expectedInitData.LastSeqNum,
+			ReceivedLastSeqNum:  expectedInitData.LastSeqNum,
+			LastSeqNumRequested: true,
 		})
 
 		initData, err := w.WaitInit(context.Background())
@@ -432,9 +434,15 @@ func TestWriterImpl_Reconnect(t *testing.T) {
 			connectionError error
 		}
 
+		isFirstConnection := true
 		newStream := func(name string) *MockRawTopicWriterStream {
 			strm := NewMockRawTopicWriterStream(mc)
 			initReq := testCreateInitRequest(w)
+			if isFirstConnection {
+				isFirstConnection = false
+			} else {
+				initReq.GetLastSeqNo = false
+			}
 
 			streamClosed := make(empty.Chan)
 			strm.EXPECT().CloseSend().Do(func() {

internal/topic/topicwriterinternal/writer_single_stream.go

@@ -24,7 +24,7 @@ type SingleStreamWriterConfig struct {
 	stream                RawTopicWriterStream
 	queue                 *messageQueue
 	encodersMap           *EncoderMap
-	getAutoSeq            bool
+	getLastSeqNum         bool
 	reconnectorInstanceID string
 }
 
@@ -33,25 +33,26 @@ func newSingleStreamWriterConfig(
 	stream RawTopicWriterStream,
 	queue *messageQueue,
 	encodersMap *EncoderMap,
-	getAutoSeq bool,
+	getLastSeqNum bool,
 	reconnectorID string,
 ) SingleStreamWriterConfig {
 	return SingleStreamWriterConfig{
 		WritersCommonConfig:   common,
 		stream:                stream,
 		queue:                 queue,
 		encodersMap:           encodersMap,
-		getAutoSeq:            getAutoSeq,
+		getLastSeqNum:         getLastSeqNum,
 		reconnectorInstanceID: reconnectorID,
 	}
 }
 
 type SingleStreamWriter struct {
-	ReceivedLastSeqNum int64
-	SessionID          string
-	PartitionID        int64
-	CodecsFromServer   rawtopiccommon.SupportedCodecs
-	Encoder            EncoderSelector
+	ReceivedLastSeqNum  int64
+	LastSeqNumRequested bool
+	SessionID           string
+	PartitionID         int64
+	CodecsFromServer    rawtopiccommon.SupportedCodecs
+	Encoder             EncoderSelector
 
 	cfg            SingleStreamWriterConfig
 	allowedCodecs  rawtopiccommon.SupportedCodecs
@@ -152,6 +153,7 @@ func (w *SingleStreamWriter) initStream() (err error) {
 	)
 
 	w.SessionID = result.SessionID
+	w.LastSeqNumRequested = req.GetLastSeqNo
 	w.ReceivedLastSeqNum = result.LastSeqNo
 	w.PartitionID = result.PartitionID
 	w.CodecsFromServer = result.SupportedCodecs
@@ -164,7 +166,7 @@ func (w *SingleStreamWriter) createInitRequest() rawtopicwriter.InitRequest {
 		ProducerID:       w.cfg.producerID,
 		WriteSessionMeta: w.cfg.writerMeta,
 		Partitioning:     w.cfg.defaultPartitioning,
-		GetLastSeqNo:     w.cfg.getAutoSeq,
+		GetLastSeqNo:     w.cfg.getLastSeqNum,
 	}
 }
 

internal/topic/topicwriterinternal/writer_single_stream_test.go

@@ -20,7 +20,7 @@ func TestWriterImpl_CreateInitMessage(t *testing.T) {
 				defaultPartitioning: rawtopicwriter.NewPartitioningPartitionID(5),
 				compressorCount:     1,
 			},
-			getAutoSeq: true,
+			getLastSeqNum: true,
 		}
 		w := newSingleStreamWriterStopped(ctx, cfg)
 		expected := rawtopicwriter.InitRequest{
@@ -36,7 +36,7 @@ func TestWriterImpl_CreateInitMessage(t *testing.T) {
 	t.Run("WithoutGetLastSeq", func(t *testing.T) {
 		ctx := xtest.Context(t)
 		w := newSingleStreamWriterStopped(ctx,
-			SingleStreamWriterConfig{getAutoSeq: false},
+			SingleStreamWriterConfig{getLastSeqNum: false},
 		)
 		require.False(t, w.createInitRequest().GetLastSeqNo)
 	})

tests/integration/topic_regression_test.go

@@ -0,0 +1,35 @@
+//go:build integration
+// +build integration
+
+package integration
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/ydb-platform/ydb-go-sdk/v3/topic/topicoptions"
+	"github.com/ydb-platform/ydb-go-sdk/v3/topic/topicwriter"
+)
+
+func TestRegressionIssue1011_WriteInitInfoLastSeqNum(t *testing.T) {
+	scope := newScope(t)
+	w1 := scope.TopicWriter()
+	err := w1.Write(scope.Ctx, topicwriter.Message{
+		Data: strings.NewReader("123"),
+	})
+	require.NoError(t, err)
+	require.NoError(t, w1.Close(scope.Ctx))
+
+	// Check
+	w2, err := scope.Driver().Topic().StartWriter(
+		scope.TopicPath(),
+		topicoptions.WithWriterProducerID(scope.TopicWriterProducerID()),
+		topicoptions.WithWriterSetAutoSeqNo(false),
+	)
+	require.NoError(t, err)
+
+	info, err := w2.WaitInitInfo(scope.Ctx)
+	require.Equal(t, int64(1), info.LastSeqNum)
+}