add custom consumer name

add consumer custom name option
commit_hash:0a98b0eacf582366c69a8719c638db00a2c82ee2

pkg/providers/ydb/model_source.go

@@ -45,10 +45,11 @@ type YdbSource struct {
 	RootCAFiles []string
 
 	// replication stuff:
-	ChangeFeedMode       ChangeFeedModeType
-	ChangeFeedCustomName string          // user can specify pre-created feed's name, otherwise it will created with name == transferID
-	BufferSize           model.BytesSize // it's not some real buffer size - see comments to waitLimits() method in kafka-source
-	VerboseSDKLogs       bool
+	ChangeFeedMode               ChangeFeedModeType
+	ChangeFeedCustomName         string // user can specify pre-created feed's name, otherwise it will created with name == transferID
+	ChangeFeedCustomConsumerName string
+	BufferSize                   model.BytesSize // it's not some real buffer size - see comments to waitLimits() method in kafka-source
+	VerboseSDKLogs               bool
 
 	// auth stuff:
 	Token            model.SecretString

pkg/providers/ydb/reader_threadsafe.go

@@ -38,7 +38,7 @@ func (r *readerThreadSafe) Close(ctx context.Context) error {
 	return r.readerImpl.Close(ctx)
 }
 
-func newReader(feedName string, dbname string, tables []string, ydbClient *ydb.Driver, logger log.Logger) (*readerThreadSafe, error) {
+func newReader(feedName, consumerName, dbname string, tables []string, ydbClient *ydb.Driver, logger log.Logger) (*readerThreadSafe, error) {
 	dbname = strings.TrimLeft(dbname, "/")
 	selectors := make([]topicoptions.ReadSelector, len(tables))
 	for i, table := range tables {
@@ -49,7 +49,7 @@ func newReader(feedName string, dbname string, tables []string, ydbClient *ydb.D
 	}
 
 	readerImpl, err := ydbClient.Topic().StartReader(
-		dataTransferConsumerName,
+		consumerName,
 		selectors,
 		topicoptions.WithReaderCommitTimeLagTrigger(0),
 		topicoptions.WithReaderCommitMode(topicoptions.CommitModeSync),

pkg/providers/ydb/source.go

@@ -266,7 +266,12 @@ func NewSource(transferID string, cfg *YdbSource, logger log.Logger, _ metrics.R
 	if cfg.ChangeFeedCustomName != "" {
 		feedName = cfg.ChangeFeedCustomName
 	}
-	reader, err := newReader(feedName, cfg.Database, cfg.Tables, ydbClient, logger)
+	consumerName := dataTransferConsumerName
+	if cfg.ChangeFeedCustomConsumerName != "" {
+		consumerName = cfg.ChangeFeedCustomConsumerName
+	}
+
+	reader, err := newReader(feedName, consumerName, cfg.Database, cfg.Tables, ydbClient, logger)
 	if err != nil {
 		return nil, xerrors.Errorf("failed to create stream reader: %w", err)
 	}