Merge pull request #1264 from ydb-platform/fix-goroutine-labels

fix goroutine labels

Author: rekby

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),
 	}
 }