resolved conflict

pingcap · May 5, 2022 · 4122dc4 · 4122dc4
1 parent 8559044
commit 4122dc4
Show file tree

Hide file tree

Showing 4 changed files with 6 additions and 9 deletions.
diff --git a/cdc/owner/changefeed.go b/cdc/owner/changefeed.go
@@ -34,7 +34,6 @@ import (
 	"github.com/pingcap/tiflow/pkg/orchestrator"
 	"github.com/pingcap/tiflow/pkg/txnutil/gc"
 	"github.com/pingcap/tiflow/pkg/upstream"
-	"github.com/pingcap/tiflow/pkg/util"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/tikv/client-go/v2/oracle"
 	"go.uber.org/zap"

diff --git a/cdc/owner/ddl_puller.go b/cdc/owner/ddl_puller.go
@@ -85,7 +85,6 @@ func newDDLPuller(ctx cdcContext.Context, upStream *upstream.Upstream, startTs u
 			kvStorage,
 			upStream.PDClock,
 			// Add "_ddl_puller" to make it different from table pullers.
-			ctx.ChangefeedVars().ID+"_ddl_puller",
 			model.ChangeFeedID{
 				Namespace: ctx.ChangefeedVars().ID.Namespace,
 				// Add "_ddl_puller" to make it different from table pullers.

diff --git a/pkg/txnutil/gc/gc_manager_test.go b/pkg/txnutil/gc/gc_manager_test.go
@@ -102,18 +102,18 @@ func (s *gcManagerSuite) TestCheckStaleCheckpointTs(c *check.C) {
 	time.Sleep(1 * time.Second)
 
 	cfID := model.DefaultChangeFeedID("cfID")
-	err = gcManager.CheckStaleCheckpointTs(cCtx, cfID, 10)
+	err := gcManager.CheckStaleCheckpointTs(ctx, cfID, 10)
 	c.Assert(cerror.ErrGCTTLExceeded.Equal(errors.Cause(err)), check.IsTrue)
 	c.Assert(cerror.ChangefeedFastFailError(err), check.IsTrue)
 
-	err = gcManager.CheckStaleCheckpointTs(cCtx, cfID, oracle.GoTimeToTS(time.Now()))
+	err = gcManager.CheckStaleCheckpointTs(ctx, cfID, oracle.GoTimeToTS(time.Now()))
 	c.Assert(err, check.IsNil)
 
 	gcManager.isTiCDCBlockGC = false
 	gcManager.lastSafePointTs = 20
 
-	err = gcManager.CheckStaleCheckpointTs(cCtx, cfID, 10)
-  
+	err = gcManager.CheckStaleCheckpointTs(ctx, cfID, 10)
+
 	c.Assert(cerror.ErrSnapshotLostByGC.Equal(errors.Cause(err)), check.IsTrue)
 	c.Assert(cerror.ChangefeedFastFailError(err), check.IsTrue)
 }
diff --git a/pkg/upstream/manager.go b/pkg/upstream/manager.go
@@ -15,11 +15,10 @@ package upstream
 
 import (
 	"context"
-	"log"
 	"sync"
-	"sync/atomic"
 
 	"github.com/pingcap/errors"
+	"github.com/pingcap/log"
 	"github.com/pingcap/tiflow/pkg/config"
 	pd "github.com/tikv/pd/client"
 	"go.uber.org/zap"
@@ -48,8 +47,8 @@ func NewManager(ctx context.Context) *Manager {
 // NewManager4Test returns a Manager for unit test.
 func NewManager4Test(pdClient pd.Client) *Manager {
 	up := NewUpstream4Test(pdClient)
-	atomic.StoreInt32(&up.status, normal)
 	res := &Manager{ups: new(sync.Map)}
+	res.ups.Store(DefaultClusterID, up)
 	return res
 }