wip

atzoum · atzoum · commit 9f9adeb04eec · 2026-01-23T22:44:12.000+02:00
🔒 Scanned for secrets using gitleaks 8.30.0
diff --git a/app/apphandlers/embeddedAppHandler.go b/app/apphandlers/embeddedAppHandler.go
@@ -247,12 +247,13 @@ func (a *embeddedApp) StartRudderCore(ctx context.Context, shutdownFn func(), op
 	}
 
 	// setup partition migrator
-	partitionMigrator, gwWODB, rtRWDB, brtRWDB, err := setupProcessorPartitionMigrator(ctx, shutdownFn, dbPool,
+	partitionMigrator, gwWODB, rtRWDB, brtRWDB, finally, err := setupProcessorPartitionMigrator(ctx, shutdownFn, dbPool,
 		config, statsFactory,
 		gwRODB, gwWODB,
 		rtRWDB, brtRWDB,
 		modeProvider.EtcdClient,
 	)
+	defer finally() // always defer finally
 	if err != nil {
 		return fmt.Errorf("setting up partition migrator: %w", err)
 	}
diff --git a/app/apphandlers/processorAppHandler.go b/app/apphandlers/processorAppHandler.go
@@ -232,12 +232,13 @@ func (a *processorApp) StartRudderCore(ctx context.Context, shutdownFn func(), o
 	}
 
 	// setup partition migrator
-	partitionMigrator, gwRODB, rtRWDB, brtRWDB, err := setupProcessorPartitionMigrator(ctx, shutdownFn, dbPool,
+	partitionMigrator, gwRODB, rtRWDB, brtRWDB, finally, err := setupProcessorPartitionMigrator(ctx, shutdownFn, dbPool,
 		config, statsFactory,
 		gwRODB, nil,
 		rtRWDB, brtRWDB,
 		modeProvider.EtcdClient,
 	)
+	defer finally() // always defer finally
 	if err != nil {
 		return fmt.Errorf("setting up partition migrator: %w", err)
 	}
diff --git a/app/apphandlers/setup_partitionmigration.go b/app/apphandlers/setup_partitionmigration.go
@@ -33,21 +33,22 @@ func setupProcessorPartitionMigrator(ctx context.Context,
 	gwRODB, gwWODB,
 	rtRWDB, brtRWDB jobsdb.JobsDB,
 	etcdClientProvider func() (etcdclient.Client, error),
-) (partitionMigrator PartitionMigrator, gwDB, rtDB, brtDB jobsdb.JobsDB, err error) {
+) (partitionMigrator PartitionMigrator, gwDB, rtDB, brtDB jobsdb.JobsDB, finally func(), err error) {
 	if !config.GetBool("PartitionMigration.enabled", false) {
+		finally = func() {}
 		if gwWODB == nil {
 			// caller expects to get reader gw db back if writer is nil
-			return &noOpPartitionMigrator{}, gwRODB, rtRWDB, brtRWDB, nil
+			return &noOpPartitionMigrator{}, gwRODB, rtRWDB, brtRWDB, finally, nil
 		} else {
 			// caller expects to get writer gw db back if writer is not nil
-			return &noOpPartitionMigrator{}, gwWODB, rtRWDB, brtRWDB, nil
+			return &noOpPartitionMigrator{}, gwWODB, rtRWDB, brtRWDB, finally, nil
 		}
 	}
 
 	log := logger.NewLogger().Child("partitionmigrator")
 	partitionCount := config.GetIntVar(0, 1, "JobsDB.partitionCount")
 	if partitionCount == 0 {
-		return nil, nil, nil, nil, fmt.Errorf("partition migrator needs partition count > 0")
+		return nil, nil, nil, nil, nil, fmt.Errorf("partition migrator needs partition count > 0")
 	}
 	bufferFlushBatchSize := config.GetReloadableIntVar(20000, 1, "PartitionMigration.bufferFlushBatchSize")
 	bufferFlushPayloadSize := config.GetReloadableInt64Var(500, bytesize.MB, "PartitionMigration.bufferFlushPayloadSize")
@@ -66,6 +67,12 @@ func setupProcessorPartitionMigrator(ctx context.Context,
 			jobsdb.WithStats(stats),
 			jobsdb.WithDBHandle(dbPool),
 		)
+		if err := gwBuffRWHandle.Start(); err != nil {
+			return nil, nil, nil, nil, finally, fmt.Errorf("starting gw buffer jobsdb handle: %w", err)
+		}
+		finally = func() {
+			gwBuffRWHandle.Stop()
+		}
 		gwSetupOpt = partitionbuffer.WithSeparateReaderAndWriterPrimaryJobsDBs(gwRODB, gwWODB, gwBuffRWHandle)
 	} else {
 		// we have only a reader gw DB, so we create a buffer with reader and
@@ -98,7 +105,7 @@ func setupProcessorPartitionMigrator(ctx context.Context,
 		partitionbuffer.WithFlushMoveTimeout(bufferFlushMoveTimeout),
 	)
 	if err != nil {
-		return nil, nil, nil, nil, fmt.Errorf("creating gw partition buffer: %w", err)
+		return nil, nil, nil, nil, finally, fmt.Errorf("creating gw partition buffer: %w", err)
 	}
 
 	// setup partition buffer for router jobsDB
@@ -120,7 +127,7 @@ func setupProcessorPartitionMigrator(ctx context.Context,
 		partitionbuffer.WithFlushMoveTimeout(bufferFlushMoveTimeout),
 	)
 	if err != nil {
-		return nil, nil, nil, nil, fmt.Errorf("creating rt partition buffer: %w", err)
+		return nil, nil, nil, nil, finally, fmt.Errorf("creating rt partition buffer: %w", err)
 	}
 
 	// setup partition buffer for batchrouter jobsDB
@@ -142,21 +149,21 @@ func setupProcessorPartitionMigrator(ctx context.Context,
 		partitionbuffer.WithFlushMoveTimeout(bufferFlushMoveTimeout),
 	)
 	if err != nil {
-		return nil, nil, nil, nil, fmt.Errorf("creating batch rt partition buffer: %w", err)
+		return nil, nil, nil, nil, finally, fmt.Errorf("creating batch rt partition buffer: %w", err)
 	}
 
 	// setup partition migrator
 	etcdClient, err := etcdClientProvider()
 	if err != nil {
-		return nil, nil, nil, nil, fmt.Errorf("getting etcd client: %w", err)
+		return nil, nil, nil, nil, finally, fmt.Errorf("getting etcd client: %w", err)
 	}
 	nodeIndex := config.GetIntVar(-1, 1, "PROCESSOR_INDEX")
 	if nodeIndex < 0 {
-		return nil, nil, nil, nil, fmt.Errorf("got invalid node index from config: %d", nodeIndex)
+		return nil, nil, nil, nil, finally, fmt.Errorf("got invalid node index from config: %d", nodeIndex)
 	}
 	nodeName := config.GetStringVar("", "HOSTNAME")
 	if nodeName == "" {
-		return nil, nil, nil, nil, fmt.Errorf("got empty node name from config")
+		return nil, nil, nil, nil, finally, fmt.Errorf("got empty node name from config")
 	}
 
 	targetURLProvider, err := func() (func(targetNodeIndex int) (string, error), error) {
@@ -177,7 +184,7 @@ func setupProcessorPartitionMigrator(ctx context.Context,
 		}, nil
 	}()
 	if err != nil {
-		return nil, nil, nil, nil, fmt.Errorf("creating target URL provider: %w", err)
+		return nil, nil, nil, nil, finally, fmt.Errorf("creating target URL provider: %w", err)
 	}
 
 	sourceMigrator, err := sourcenode.NewMigratorBuilder(nodeIndex, nodeName).
@@ -190,7 +197,7 @@ func setupProcessorPartitionMigrator(ctx context.Context,
 		WithReaderJobsDBs([]jobsdb.JobsDB{gwRODB, rtRWDB, brtRWDB}).
 		Build()
 	if err != nil {
-		return nil, nil, nil, nil, fmt.Errorf("creating source node migrator: %w", err)
+		return nil, nil, nil, nil, finally, fmt.Errorf("creating source node migrator: %w", err)
 	}
 
 	targetMigrator, err := targetnode.NewMigratorBuilder(nodeIndex, nodeName).
@@ -205,7 +212,7 @@ func setupProcessorPartitionMigrator(ctx context.Context,
 		WithUnbufferedJobsDBs([]jobsdb.JobsDB{rtRWDB, brtRWDB, gwWODB}).
 		Build()
 	if err != nil {
-		return nil, nil, nil, nil, fmt.Errorf("creating target node migrator: %w", err)
+		return nil, nil, nil, nil, finally, fmt.Errorf("creating target node migrator: %w", err)
 	}
 	ppm, err := processor.NewProcessorPartitionMigratorBuilder(nodeIndex, nodeName).
 		WithConfig(config).
@@ -216,9 +223,9 @@ func setupProcessorPartitionMigrator(ctx context.Context,
 		WithTargetMigrator(targetMigrator).
 		Build()
 	if err != nil {
-		return nil, nil, nil, nil, fmt.Errorf("creating processor partition migrator: %w", err)
+		return nil, nil, nil, nil, finally, fmt.Errorf("creating processor partition migrator: %w", err)
 	}
-	return ppm, gwPartitionBuffer, rtPartitionBuffer, brtPartitionBuffer, nil
+	return ppm, gwPartitionBuffer, rtPartitionBuffer, brtPartitionBuffer, finally, nil
 }
 
 func setupGatewayPartitionMigrator(_ context.Context, _ *config.Config, gwWODB jobsdb.JobsDB,
diff --git a/cluster/migrator/partitionmigration/client/migration_job_executor.go b/cluster/migrator/partitionmigration/client/migration_job_executor.go
@@ -323,7 +323,7 @@ func (mpe *migrationJobExecutor) Run(ctx context.Context) error {
 			case <-done:
 				return nil
 			case <-time.After(mpe.progressPeriod.Load()):
-				mpe.logger.Infon("Partition migration in progress",
+				mpe.logger.Infon("Partition migration job in progress",
 					logger.NewIntField("sent", totalSent.Load()),
 					logger.NewIntField("acked", totalAcked.Load()),
 				)
@@ -335,7 +335,7 @@ func (mpe *migrationJobExecutor) Run(ctx context.Context) error {
 		defer close(done)
 		if err := streamGroup.Wait(); err != nil {
 			if ctx.Err() == nil {
-				mpe.logger.Errorn("Partition migration failed", obskit.Error(err))
+				mpe.logger.Errorn("Partition migration job failed", obskit.Error(err))
 				for index, jobs := range unackedBatches { // no need to lock unackedBatches, we are done with sender and receiver goroutines
 					if err := mpe.updateJobStatus(ctx, jobs, jobsdb.Failed.State, fmt.Errorf("job migration interrupted: %w", err)); err != nil {
 						mpe.logger.Warnn("Could not mark non-migrated jobs as failed",
@@ -350,13 +350,13 @@ func (mpe *migrationJobExecutor) Run(ctx context.Context) error {
 			}
 			return fmt.Errorf("migrating partitions: %w", err)
 		}
-		mpe.logger.Infon("Partition migration completed successfully",
+		mpe.logger.Infon("Partition migration job completed successfully",
 			logger.NewIntField("total", totalAcked.Load()),
 		)
 		return nil
 	})
 	err = g.Wait()
-	mpe.logger.Infon("Partition migration progress final status",
+	mpe.logger.Infon("Partition migration job progress final status",
 		logger.NewIntField("sent", totalSent.Load()),
 		logger.NewIntField("acked", totalAcked.Load()),
 		obskit.Error(err),
diff --git a/cluster/migrator/processor/sourcenode/sourcenode_migrator.go b/cluster/migrator/processor/sourcenode/sourcenode_migrator.go
@@ -166,7 +166,7 @@ func (m *migrator) waitForNoInProgressJobs(ctx context.Context, sourcePartitions
 							logger.NewDurationField("elapsed", time.Since(start)),
 						)
 						// sleep for a short duration before checking again
-						if err := misc.SleepCtx(pollCtx, 5*time.Second); err != nil {
+						if err := misc.SleepCtx(pollCtx, m.c.inProgressPollSleep.Load()); err != nil {
 							return fmt.Errorf("sleeping while waiting for no in-progress jobs in %q jobsdb: %w", readerJobsDB.Identifier(), err)
 						}
 					}
@@ -322,7 +322,7 @@ func (m *migrator) onNewJob(ctx context.Context, key string, job *etcdtypes.Part
 			delete(m.pendingMigrationJobs, job.JobID)
 			m.pendingMigrationJobsMu.Unlock()
 
-			log.Infon("Partition migration job status marked as moved successfully",
+			log.Infon("Partition migration job status marked as moved in etcd successfully",
 				logger.NewIntField("revision", res.Header.Revision),
 			)
 			m.stats.NewTaggedStat("partition_mig_src_job", stats.TimerType, m.statsTags()).SendTiming(time.Since(start))
diff --git a/cluster/migrator/processor/sourcenode/sourcenode_migrator_builder.go b/cluster/migrator/processor/sourcenode/sourcenode_migrator_builder.go
@@ -116,9 +116,9 @@ func (b *MigratorBuilder) Build() (Migrator, error) {
 		targetURLProvider:    b.targetURLProvider,
 		pendingMigrationJobs: map[string]struct{}{},
 	}
-	m.c.readExcludeSleep = b.config.GetReloadableDurationVar(20, time.Second, "PartitionMigration.Processor.SourceNode.readExcludeSleep")
+	m.c.readExcludeSleep = b.config.GetReloadableDurationVar(30, time.Second, "PartitionMigration.Processor.SourceNode.readExcludeSleep")
 	m.c.waitForInProgressTimeout = b.config.GetReloadableDurationVar(5, time.Minute, "PartitionMigration.Processor.SourceNode.waitForInProgressTimeout")
-	m.c.inProgressPollSleep = b.config.GetReloadableDurationVar(1, time.Second, "PartitionMigration.SourceNode.inProgressPollSleep")
+	m.c.inProgressPollSleep = b.config.GetReloadableDurationVar(5, time.Second, "PartitionMigration.SourceNode.inProgressPollSleep")
 
 	return m, nil
 }
diff --git a/cluster/migrator/processor/targetnode/targetnode_migrator.go b/cluster/migrator/processor/targetnode/targetnode_migrator.go
@@ -231,7 +231,7 @@ func (m *migrator) onNewJob(ctx context.Context, key string, job *etcdtypes.Part
 			delete(m.pendingMigrationJobs, job.JobID)
 			m.pendingMigrationJobsMu.Unlock()
 
-			log.Infon("Partition migration job status marked as completed successfully",
+			log.Infon("Partition migration job status marked as completed in etcd successfully",
 				logger.NewIntField("revision", res.Header.Revision),
 			)
 			m.stats.NewTaggedStat("partition_mig_target_job", stats.TimerType, m.statsTags()).SendTiming(time.Since(start))
diff --git a/cluster/partitionbuffer/jobsdb_partition_buffer_factory.go b/cluster/partitionbuffer/jobsdb_partition_buffer_factory.go
@@ -69,7 +69,7 @@ func WithSeparateReaderAndWriterPrimaryJobsDBs(primaryReader, primaryWriter, buf
 		b.canStore = true
 		b.canFlush = true
 		b.differentBufferDBs = false
-		b.lifecycleJobsDBs = []jobsdb.JobsDB{primaryReader, buffer} // primaryWriter's lifecycle is externally managed
+		b.lifecycleJobsDBs = []jobsdb.JobsDB{primaryReader} // primaryWriter's and buffer's lifecycle are externally managed
 	}
 }
 
diff --git a/cluster/partitionbuffer/jobsdb_partition_buffer_flush.go b/cluster/partitionbuffer/jobsdb_partition_buffer_flush.go
@@ -63,7 +63,7 @@ func (b *jobsDBPartitionBuffer) FlushBufferedPartitions(ctx context.Context, par
 	var totalCount int
 
 	// move in batches until we stop reaching limits
-	b.logger.Infon("flushing jobs from buffer to primary jobsdb (move phase)",
+	b.logger.Infon("Flushing jobs from buffer to primary jobsdb (move phase)",
 		logger.NewStringField("partitions", strings.Join(partitions, ",")),
 		logger.NewStringField("prefix", b.Identifier()),
 	)
@@ -72,7 +72,7 @@ func (b *jobsDBPartitionBuffer) FlushBufferedPartitions(ctx context.Context, par
 		select {
 		case <-moveTimeout:
 			// timeout reached, break out to switchover
-			b.logger.Warnn("flush move timeout reached, proceeding to switchover",
+			b.logger.Warnn("Flush move timeout reached, proceeding to switchover",
 				logger.NewStringField("partitions", fmt.Sprintf("%v", partitions)),
 				logger.NewDurationField("duration", time.Since(start)),
 			)
@@ -87,7 +87,7 @@ func (b *jobsDBPartitionBuffer) FlushBufferedPartitions(ctx context.Context, par
 		}
 	}
 	// switchover
-	b.logger.Infon("flushing jobs from buffer to primary jobsdb (switchover phase)",
+	b.logger.Infon("Flushing jobs from buffer to primary jobsdb (switchover phase)",
 		logger.NewStringField("partitions", strings.Join(partitions, ",")),
 		logger.NewStringField("prefix", b.Identifier()),
 	)
@@ -96,7 +96,7 @@ func (b *jobsDBPartitionBuffer) FlushBufferedPartitions(ctx context.Context, par
 		return fmt.Errorf("switchover of buffered partitions: %w", err)
 	}
 	totalCount += switchoverCount
-	b.logger.Infon("completed flush of buffered partitions",
+	b.logger.Infon("Flushing of buffered partitions completed successfully",
 		logger.NewStringField("partitions", strings.Join(partitions, ",")),
 		logger.NewStringField("prefix", b.Identifier()),
 		logger.NewDurationField("duration", time.Since(start)),
@@ -164,20 +164,16 @@ func (b *jobsDBPartitionBuffer) switchoverBufferedPartitions(ctx context.Context
 		"prefix": b.Identifier(),
 	}).RecordDuration()()
 
-	lockCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
-	if !b.bufferedPartitionsMu.TryLockWithContext(lockCtx) {
-		cancel()
-		// TODO: get a goroutine dump to /Users/aristzoumas/Desktop/lock.dump
+	if !b.bufferedPartitionsMu.TryLockWithContext(ctx) {
 		return 0, fmt.Errorf("acquiring a buffered partitions write lock during switchover: %w", ctx.Err())
 	}
-	cancel()
-	b.logger.Infon("buffered partitions write lock acquired (switchover phase)",
+	b.logger.Infon("Buffered partitions write lock acquired (switchover phase)",
 		logger.NewStringField("partitions", strings.Join(partitionIDs, ",")),
 		logger.NewStringField("prefix", b.Identifier()),
 	)
 	defer func() {
 		b.bufferedPartitionsMu.Unlock()
-		b.logger.Infon("buffered partitions write lock released (switchover phase)",
+		b.logger.Infon("Buffered partitions write lock released (switchover phase)",
 			logger.NewStringField("partitions", strings.Join(partitionIDs, ",")),
 			logger.NewStringField("prefix", b.Identifier()),
 		)
diff --git a/integration_test/partitionmigration/client_test.go b/integration_test/partitionmigration/client_test.go
@@ -14,9 +14,10 @@ import (
 	"time"
 
 	"github.com/google/uuid"
+	"golang.org/x/sync/errgroup"
+
 	"github.com/rudderlabs/rudder-go-kit/partmap"
 	"github.com/rudderlabs/rudder-server/utils/httputil"
-	"golang.org/x/sync/errgroup"
 )
 
 type gatewayClientConfig struct {
@@ -96,7 +97,7 @@ func (gc *gatewayClient) GetTotalSent() int64 {
 	return gc.totalSent.Load()
 }
 
-func (gc *gatewayClient) generateJobs(userID string, startIndex int, jobCount int) (jobs []string) {
+func (gc *gatewayClient) generateJobs(userID string, startIndex, jobCount int) (jobs []string) {
 	for j := startIndex; j < startIndex+jobCount; j++ {
 		messageID := uuid.New().String()
 		timestamp := time.Now().UTC().Format("2006-01-02T15:04:05.000Z")
@@ -114,23 +115,23 @@ func (gc *gatewayClient) generateJobs(userID string, startIndex int, jobCount in
 }
 
 func (gc *gatewayClient) getUserIDForPartition(partitionIdx, numPartitions int) string {
-	const separator = "<<>>"
 	for i := range 1000000 {
 		candidate := "user" + strconv.Itoa(i)
-		idx, _ := partmap.Murmur3Partition32(gwUserID(candidate), uint32(numPartitions))
+		idx, _ := partmap.Murmur3Partition32(legacyUserID(candidate), uint32(numPartitions))
 		if idx == uint32(partitionIdx) {
 			return candidate
 		}
 	}
 	return ""
 }
 
-func gwUserID(userID string) string {
+// gateway component doesn't hash just the userID, but a weird combination of separators and ids of some sort
+func legacyUserID(userID string) string {
 	const separator = "<<>>"
 	return separator + userID + separator + userID
 }
 
-func (gc *gatewayClient) sendRequest(urlString, writeKey string, partitionKey, payload string) error {
+func (gc *gatewayClient) sendRequest(urlString, writeKey, userID, payload string) error {
 	u, _ := url.Parse(urlString)
 	u.Path = path.Join(u.Path, "v1", "identify")
 	requestURL := u.String()
@@ -140,7 +141,7 @@ func (gc *gatewayClient) sendRequest(urlString, writeKey string, partitionKey, p
 	}
 	req.Header.Add("Content-Type", "application/json")
 	req.Header.Add("Authorization", "Basic "+base64.StdEncoding.EncodeToString([]byte(writeKey+":")))
-	req.Header.Add("X-Partition-Key", gwUserID(partitionKey))
+	req.Header.Add("X-Partition-Key", legacyUserID(userID))
 	res, err := http.DefaultClient.Do(req)
 	defer func() { httputil.CloseResponse(res) }()
 	if err != nil {
diff --git a/integration_test/partitionmigration/partitionmigration_embedded_test.go b/integration_test/partitionmigration/partitionmigration_embedded_test.go
diff --git a/integration_test/partitionmigration/rudderserver_test.go b/integration_test/partitionmigration/rudderserver_test.go
diff --git a/testhelper/clustertest/migrationexecutor.go b/testhelper/clustertest/migrationexecutor.go
diff --git a/testhelper/clustertest/routingproxy.go b/testhelper/clustertest/routingproxy.go

Original file line number	Diff line number	Diff line change
`@@ -247,12 +247,13 @@ func (a *embeddedApp) StartRudderCore(ctx context.Context, shutdownFn func(), op`
`247`	`247`	`}`
`248`	`248`
`249`	`249`	`// setup partition migrator`
`250`		`- partitionMigrator, gwWODB, rtRWDB, brtRWDB, err := setupProcessorPartitionMigrator(ctx, shutdownFn, dbPool,`
	`250`	`+ partitionMigrator, gwWODB, rtRWDB, brtRWDB, finally, err := setupProcessorPartitionMigrator(ctx, shutdownFn, dbPool,`
`251`	`251`	`config, statsFactory,`
`252`	`252`	`gwRODB, gwWODB,`
`253`	`253`	`rtRWDB, brtRWDB,`
`254`	`254`	`modeProvider.EtcdClient,`
`255`	`255`	`)`
	`256`	`+ defer finally() // always defer finally`
`256`	`257`	`if err != nil {`
`257`	`258`	`return fmt.Errorf("setting up partition migrator: %w", err)`
`258`	`259`	`}`
Original file line number	Diff line number	Diff line change
`@@ -232,12 +232,13 @@ func (a *processorApp) StartRudderCore(ctx context.Context, shutdownFn func(), o`
`232`	`232`	`}`
`233`	`233`
`234`	`234`	`// setup partition migrator`
`235`		`- partitionMigrator, gwRODB, rtRWDB, brtRWDB, err := setupProcessorPartitionMigrator(ctx, shutdownFn, dbPool,`
	`235`	`+ partitionMigrator, gwRODB, rtRWDB, brtRWDB, finally, err := setupProcessorPartitionMigrator(ctx, shutdownFn, dbPool,`
`236`	`236`	`config, statsFactory,`
`237`	`237`	`gwRODB, nil,`
`238`	`238`	`rtRWDB, brtRWDB,`
`239`	`239`	`modeProvider.EtcdClient,`
`240`	`240`	`)`
	`241`	`+ defer finally() // always defer finally`
`241`	`242`	`if err != nil {`
`242`	`243`	`return fmt.Errorf("setting up partition migrator: %w", err)`
`243`	`244`	`}`
Original file line number	Diff line number	Diff line change
`@@ -166,7 +166,7 @@ func (m *migrator) waitForNoInProgressJobs(ctx context.Context, sourcePartitions`
`166`	`166`	`logger.NewDurationField("elapsed", time.Since(start)),`
`167`	`167`	`)`
`168`	`168`	`// sleep for a short duration before checking again`
`169`		`- if err := misc.SleepCtx(pollCtx, 5*time.Second); err != nil {`
	`169`	`+ if err := misc.SleepCtx(pollCtx, m.c.inProgressPollSleep.Load()); err != nil {`
`170`	`170`	`return fmt.Errorf("sleeping while waiting for no in-progress jobs in %q jobsdb: %w", readerJobsDB.Identifier(), err)`
`171`	`171`	`}`
`172`	`172`	`}`
`@@ -322,7 +322,7 @@ func (m migrator) onNewJob(ctx context.Context, key string, job etcdtypes.Part`
`322`	`322`	`delete(m.pendingMigrationJobs, job.JobID)`
`323`	`323`	`m.pendingMigrationJobsMu.Unlock()`
`324`	`324`
`325`		`- log.Infon("Partition migration job status marked as moved successfully",`
	`325`	`+ log.Infon("Partition migration job status marked as moved in etcd successfully",`
`326`	`326`	`logger.NewIntField("revision", res.Header.Revision),`
`327`	`327`	`)`
`328`	`328`	`m.stats.NewTaggedStat("partition_mig_src_job", stats.TimerType, m.statsTags()).SendTiming(time.Since(start))`
Original file line number	Diff line number	Diff line change
`@@ -116,9 +116,9 @@ func (b *MigratorBuilder) Build() (Migrator, error) {`
`116`	`116`	`targetURLProvider: b.targetURLProvider,`
`117`	`117`	`pendingMigrationJobs: map[string]struct{}{},`
`118`	`118`	`}`
`119`		`- m.c.readExcludeSleep = b.config.GetReloadableDurationVar(20, time.Second, "PartitionMigration.Processor.SourceNode.readExcludeSleep")`
	`119`	`+ m.c.readExcludeSleep = b.config.GetReloadableDurationVar(30, time.Second, "PartitionMigration.Processor.SourceNode.readExcludeSleep")`
`120`	`120`	`m.c.waitForInProgressTimeout = b.config.GetReloadableDurationVar(5, time.Minute, "PartitionMigration.Processor.SourceNode.waitForInProgressTimeout")`
`121`		`- m.c.inProgressPollSleep = b.config.GetReloadableDurationVar(1, time.Second, "PartitionMigration.SourceNode.inProgressPollSleep")`
	`121`	`+ m.c.inProgressPollSleep = b.config.GetReloadableDurationVar(5, time.Second, "PartitionMigration.SourceNode.inProgressPollSleep")`
`122`	`122`
`123`	`123`	`return m, nil`
`124`	`124`	`}`
Original file line number	Diff line number	Diff line change
`@@ -231,7 +231,7 @@ func (m migrator) onNewJob(ctx context.Context, key string, job etcdtypes.Part`
`231`	`231`	`delete(m.pendingMigrationJobs, job.JobID)`
`232`	`232`	`m.pendingMigrationJobsMu.Unlock()`
`233`	`233`
`234`		`- log.Infon("Partition migration job status marked as completed successfully",`
	`234`	`+ log.Infon("Partition migration job status marked as completed in etcd successfully",`
`235`	`235`	`logger.NewIntField("revision", res.Header.Revision),`
`236`	`236`	`)`
`237`	`237`	`m.stats.NewTaggedStat("partition_mig_target_job", stats.TimerType, m.statsTags()).SendTiming(time.Since(start))`
Original file line number	Diff line number	Diff line change
`@@ -69,7 +69,7 @@ func WithSeparateReaderAndWriterPrimaryJobsDBs(primaryReader, primaryWriter, buf`
`69`	`69`	`b.canStore = true`
`70`	`70`	`b.canFlush = true`
`71`	`71`	`b.differentBufferDBs = false`
`72`		`- b.lifecycleJobsDBs = []jobsdb.JobsDB{primaryReader, buffer} // primaryWriter's lifecycle is externally managed`
	`72`	`+ b.lifecycleJobsDBs = []jobsdb.JobsDB{primaryReader} // primaryWriter's and buffer's lifecycle are externally managed`
`73`	`73`	`}`
`74`	`74`	`}`
`75`	`75`