Merge pull request #1723 add flag for manage split-merge support

Author: rekby

CHANGELOG.md

@@ -1,3 +1,4 @@
+* Added option WithReaderSupportSplitMergePartitions for topic manage support of split-merge partitions on client side (enabled by default).
 * Allowed overflow queue limit for one goroutine at time for topic writer
 * Removed delay before send commit in sync mode of a topic reader
 

internal/topic/topiclistenerinternal/stream_listener.go

@@ -174,7 +174,7 @@ func (l *streamListener) initStream(ctx context.Context, client TopicClient) err
 	}
 	l.stream = topicreadercommon.NewSyncedStream(stream)
 
-	initMessage := topicreadercommon.CreateInitMessage(l.cfg.Consumer, l.cfg.Selectors)
+	initMessage := topicreadercommon.CreateInitMessage(l.cfg.Consumer, false, l.cfg.Selectors)
 	err = stream.Send(initMessage)
 	if err != nil {
 		return xerrors.WithStackTrace(fmt.Errorf("ydb: failed to send init request for read stream in the listener: %w", err))

internal/topic/topicreadercommon/init_message.go

@@ -2,10 +2,14 @@ package topicreadercommon
 
 import "github.com/ydb-platform/ydb-go-sdk/v3/internal/grpcwrapper/rawtopic/rawtopicreader"
 
-func CreateInitMessage(consumer string, selectors []*PublicReadSelector) *rawtopicreader.InitRequest {
+func CreateInitMessage(
+	consumer string,
+	supportAutoPartition bool,
+	selectors []*PublicReadSelector,
+) *rawtopicreader.InitRequest {
 	res := &rawtopicreader.InitRequest{
 		Consumer:                consumer,
-		AutoPartitioningSupport: true,
+		AutoPartitioningSupport: supportAutoPartition,
 	}
 
 	res.TopicsReadSettings = make([]rawtopicreader.TopicReadSettings, len(selectors))

internal/topic/topicreaderinternal/stream_reader_impl.go

@@ -80,18 +80,20 @@ type topicStreamReaderConfig struct {
 	GetPartitionStartOffsetCallback PublicGetPartitionStartOffsetFunc
 	CommitMode                      topicreadercommon.PublicCommitMode
 	Decoders                        topicreadercommon.DecoderMap
+	EnableSplitMergeSupport         bool
 }
 
 func newTopicStreamReaderConfig() topicStreamReaderConfig {
 	return topicStreamReaderConfig{
-		BaseContext:           context.Background(),
-		BufferSizeProtoBytes:  topicreadercommon.DefaultBufferSize,
-		Cred:                  credentials.NewAnonymousCredentials(),
-		CredUpdateInterval:    time.Hour,
-		CommitMode:            topicreadercommon.CommitModeAsync,
-		CommitterBatchTimeLag: time.Second,
-		Decoders:              topicreadercommon.NewDecoderMap(),
-		Trace:                 &trace.Topic{},
+		BaseContext:             context.Background(),
+		BufferSizeProtoBytes:    topicreadercommon.DefaultBufferSize,
+		Cred:                    credentials.NewAnonymousCredentials(),
+		CredUpdateInterval:      time.Hour,
+		CommitMode:              topicreadercommon.CommitModeAsync,
+		CommitterBatchTimeLag:   time.Second,
+		Decoders:                topicreadercommon.NewDecoderMap(),
+		Trace:                   &trace.Topic{},
+		EnableSplitMergeSupport: true,
 	}
 }
 
@@ -611,7 +613,7 @@ func (r *topicStreamReaderImpl) setStarted() error {
 }
 
 func (r *topicStreamReaderImpl) initSession() (err error) {
-	initMessage := topicreadercommon.CreateInitMessage(r.cfg.Consumer, r.cfg.ReadSelectors)
+	initMessage := topicreadercommon.CreateInitMessage(r.cfg.Consumer, r.cfg.EnableSplitMergeSupport, r.cfg.ReadSelectors)
 
 	onDone := trace.TopicOnReaderInit(r.cfg.Trace, r.readConnectionID, initMessage)
 	defer func() {

topic/topicoptions/topicoptions_reader.go

@@ -281,3 +281,11 @@ func WithReaderWithoutConsumer(saveStateOnReconnection bool) ReaderOption {
 		cfg.CommitMode = CommitModeNone
 	}
 }
+
+// WithReaderSupportSplitMergePartitions set support of split and merge partitions on client side.
+// Default is true, set false for disable the support.
+func WithReaderSupportSplitMergePartitions(enableSupport bool) ReaderOption {
+	return func(cfg *topicreaderinternal.ReaderConfig) {
+		cfg.EnableSplitMergeSupport = enableSupport
+	}
+}