Merge branch 'master' into make-some-options-public

Author: asmyasnikov

internal/query/options/retry.go

@@ -8,11 +8,11 @@ import (
 )
 
 var (
-	_ DoOption = retryOptionsOption(nil)
-	_ DoOption = traceOption{}
+	_ DoOption = RetryOptionsOption(nil)
+	_ DoOption = TraceOption{}
 
-	_ DoTxOption = retryOptionsOption(nil)
-	_ DoTxOption = traceOption{}
+	_ DoTxOption = RetryOptionsOption(nil)
+	_ DoTxOption = TraceOption{}
 	_ DoTxOption = doTxSettingsOption{}
 )
 
@@ -35,8 +35,8 @@ type (
 		txSettings tx.Settings
 	}
 
-	retryOptionsOption []retry.Option
-	traceOption        struct {
+	RetryOptionsOption []retry.Option
+	TraceOption        struct {
 		t *trace.Query
 	}
 	doTxSettingsOption struct {
@@ -56,19 +56,19 @@ func (s *doTxSettings) TxSettings() tx.Settings {
 	return s.txSettings
 }
 
-func (opt traceOption) applyDoOption(s *doSettings) {
+func (opt TraceOption) applyDoOption(s *doSettings) {
 	s.trace = s.trace.Compose(opt.t)
 }
 
-func (opt traceOption) applyDoTxOption(s *doTxSettings) {
+func (opt TraceOption) applyDoTxOption(s *doTxSettings) {
 	opt.applyDoOption(&s.doSettings)
 }
 
-func (opts retryOptionsOption) applyDoOption(s *doSettings) {
+func (opts RetryOptionsOption) applyDoOption(s *doSettings) {
 	s.retryOpts = append(s.retryOpts, opts...)
 }
 
-func (opts retryOptionsOption) applyDoTxOption(s *doTxSettings) {
+func (opts RetryOptionsOption) applyDoTxOption(s *doTxSettings) {
 	opts.applyDoOption(&s.doSettings)
 }
 
@@ -80,19 +80,19 @@ func WithTxSettings(txSettings tx.Settings) doTxSettingsOption {
 	return doTxSettingsOption{txSettings: txSettings}
 }
 
-func WithIdempotent() retryOptionsOption {
+func WithIdempotent() RetryOptionsOption {
 	return []retry.Option{retry.WithIdempotent(true)}
 }
 
-func WithLabel(lbl string) retryOptionsOption {
+func WithLabel(lbl string) RetryOptionsOption {
 	return []retry.Option{retry.WithLabel(lbl)}
 }
 
-func WithTrace(t *trace.Query) traceOption {
-	return traceOption{t: t}
+func WithTrace(t *trace.Query) TraceOption {
+	return TraceOption{t: t}
 }
 
-func WithRetryBudget(b budget.Budget) retryOptionsOption {
+func WithRetryBudget(b budget.Budget) RetryOptionsOption {
 	return []retry.Option{retry.WithBudget(b)}
 }
 

internal/query/scanner/indexed_test.go

@@ -399,7 +399,7 @@ func TestIndexed(t *testing.T) {
 				{func(v uint64) *uint64 { return &v }(100500)},
 				{func(v int64) *int64 { return &v }(100500)},
 				{func(v int32) *int32 { return &v }(100500)},
-				{func(v time.Time) *time.Time { return &v }(time.Unix(8683200000, 0))},
+				{func(v time.Time) *time.Time { return &v }(time.Unix(8683200000, 0).UTC())},
 			},
 		},
 		{

internal/query/scanner/named_test.go

@@ -411,7 +411,7 @@ func TestNamed(t *testing.T) {
 				{func(v uint64) *uint64 { return &v }(100500)},
 				{func(v int64) *int64 { return &v }(100500)},
 				{func(v int32) *int32 { return &v }(100500)},
-				{func(v time.Time) *time.Time { return &v }(time.Unix(8683200000, 0))},
+				{func(v time.Time) *time.Time { return &v }(time.Unix(8683200000, 0).UTC())},
 			},
 		},
 		{

internal/query/scanner/struct_test.go

@@ -477,7 +477,7 @@ func TestStruct(t *testing.T) {
 		Int8Int32:     123,
 		Int8Int16:     123,
 		BoolBool:      true,
-		DateTime:      time.Unix(8683200000, 0),
+		DateTime:      time.Unix(8683200000, 0).UTC(),
 		DatetimeTime:  time.Unix(100500, 0),
 		TimestampTime: time.Unix(12345678987, 654321000),
 	}, dst)

internal/topic/topicreaderinternal/committer.go

@@ -3,7 +3,6 @@ package topicreaderinternal
 import (
 	"context"
 	"errors"
-	"fmt"
 	"sync/atomic"
 	"time"
 
@@ -53,12 +52,17 @@ type committer struct {
 	commits CommitRanges
 }
 
-func newCommitterStopped(tracer *trace.Topic, lifeContext context.Context, mode PublicCommitMode, send sendMessageToServerFunc, readerID int64) *committer { //nolint:lll,revive
+func newCommitterStopped(
+	tracer *trace.Topic,
+	lifeContext context.Context, //nolint:revive
+	mode PublicCommitMode,
+	send sendMessageToServerFunc,
+) *committer {
 	res := &committer{
 		mode:             mode,
 		clock:            clockwork.NewRealClock(),
 		send:             send,
-		backgroundWorker: *background.NewWorker(lifeContext, fmt.Sprintf("ydb-topic-reader-committer: %v", readerID)),
+		backgroundWorker: *background.NewWorker(lifeContext, "ydb-topic-reader-committer"),
 		tracer:           tracer,
 	}
 	res.initChannels()

internal/topic/topicreaderinternal/committer_test.go

@@ -379,7 +379,7 @@ func TestCommitterBuffer(t *testing.T) {
 func newTestCommitter(ctx context.Context, t testing.TB) *committer {
 	res := newCommitterStopped(&trace.Topic{}, ctx, CommitModeAsync, func(msg rawtopicreader.ClientMessage) error {
 		return nil
-	}, -1)
+	})
 	res.Start()
 	t.Cleanup(func() {
 		if err := res.Close(ctx, errors.New("test committer closed")); err != nil {

internal/topic/topicreaderinternal/stream_reader_impl.go

@@ -156,12 +156,9 @@ func newTopicStreamReaderStopped(
 		rawMessagesFromBuffer: make(chan rawtopicreader.ServerMessage, 1),
 	}
 
-	res.backgroundWorkers = *background.NewWorker(stopPump, fmt.Sprintf(
-		"topic-reader-stream-background: %v",
-		res.readerID,
-	))
+	res.backgroundWorkers = *background.NewWorker(stopPump, "topic-reader-stream-background")
 
-	res.committer = newCommitterStopped(cfg.Trace, labeledContext, cfg.CommitMode, res.send, res.readerID)
+	res.committer = newCommitterStopped(cfg.Trace, labeledContext, cfg.CommitMode, res.send)
 	res.committer.BufferTimeLagTrigger = cfg.CommitterBatchTimeLag
 	res.committer.BufferCountTrigger = cfg.CommitterBatchCounterTrigger
 	res.sessionController.init()

internal/topic/topicwriterinternal/writer_single_stream.go

@@ -84,10 +84,10 @@ func newSingleStreamWriterStopped(
 ) *SingleStreamWriter {
 	return &SingleStreamWriter{
 		cfg: cfg,
-		background: *background.NewWorker(xcontext.ValueOnly(ctxForPProfLabelsOnly), fmt.Sprintf(
-			"ydb-topic-stream-writer-background: %v",
-			cfg.reconnectorInstanceID,
-		)),
+		background: *background.NewWorker(
+			xcontext.ValueOnly(ctxForPProfLabelsOnly),
+			"ydb-topic-stream-writer-background",
+		),
 		closeCompleted: make(empty.Chan),
 	}
 }