*: fix golangci-lint issues and delete unused code

.golangci.yml

@@ -67,10 +67,12 @@ linters:
         - G104
 
     # ST1000: don't require package comments.
+    # ST1003: don't enforce naming conventions on legacy identifiers.
     staticcheck:
       checks:
         - all
         - -ST1000
+        - -ST1003
 
   exclusions:
     # Strict: exclude all known generated file patterns.

Makefile

@@ -1,6 +1,6 @@
 # Phony targets are targets that are not associated with files.
 # Add new phony targets here to make them available in the `make` command.
-.PHONY: clean fmt check check-static tidy \
+.PHONY: clean fmt check check-static local-static-check tidy \
 	generate-protobuf generate_mock \
 	cdc kafka_consumer storage_consumer pulsar_consumer filter_helper \
 	prepare_test_binaries \
@@ -320,6 +320,14 @@ else
 	tools/bin/golangci-lint run --timeout 10m0s
 endif
 
+# Lint only code changed on the current branch (vs upstream/master by default).
+# Override base with LINT_BASE=<ref>.
+#   make local-static-check
+#   make local-static-check LINT_BASE=HEAD~3
+local-static-check: tools/bin/golangci-lint
+	$(eval BASE := $(if $(LINT_BASE),$(LINT_BASE),upstream/master))
+	tools/bin/golangci-lint run --timeout 10m0s --new-from-rev=$(BASE)
+
 check-ticdc-dashboard:
 	@echo "check-ticdc-dashboard"
 	@./scripts/check-ticdc-dashboard.sh

api/v2/model.go

@@ -24,7 +24,6 @@ import (
 	"github.com/pingcap/ticdc/pkg/errors"
 	"github.com/pingcap/ticdc/pkg/integrity"
 	"github.com/pingcap/ticdc/pkg/liveness"
-	"github.com/pingcap/ticdc/pkg/security"
 	"github.com/pingcap/ticdc/pkg/util"
 )
 
@@ -85,12 +84,6 @@ type VerifyTableConfig struct {
 	SinkURI       string         `json:"sink_uri"`
 }
 
-func getDefaultVerifyTableConfig() *VerifyTableConfig {
-	return &VerifyTableConfig{
-		ReplicaConfig: GetDefaultReplicaConfig(),
-	}
-}
-
 // ResumeChangefeedConfig is used by resume changefeed api
 type ResumeChangefeedConfig struct {
 	PDConfig
@@ -1350,18 +1343,6 @@ type SyncedStatus struct {
 	Info             string       `json:"info"`
 }
 
-// toCredential generates a security.Credential from a PDConfig
-func (cfg *PDConfig) toCredential() *security.Credential {
-	credential := &security.Credential{
-		CAPath:   cfg.CAPath,
-		CertPath: cfg.CertPath,
-		KeyPath:  cfg.KeyPath,
-	}
-	credential.CertAllowedCN = make([]string, len(cfg.CertAllowedCN))
-	copy(credential.CertAllowedCN, cfg.CertAllowedCN)
-	return credential
-}
-
 // Marshal returns the json marshal format of a ChangeFeedInfo
 func (info *ChangeFeedInfo) Marshal() (string, error) {
 	data, err := json.Marshal(info)

cmd/cdc/cli/cli_changefeed_helper.go

@@ -14,9 +14,7 @@
 package cli
 
 import (
-	"bufio"
 	"fmt"
-	"os"
 	"strings"
 	"time"
 
@@ -32,15 +30,6 @@ const (
 	tsGapWarning = 86400 * 1000
 )
 
-func readInput() (string, error) {
-	reader := bufio.NewReader(os.Stdin)
-	msg, err := reader.ReadString('\n')
-	if err != nil {
-		return "", err
-	}
-	return strings.TrimSpace(msg), nil
-}
-
 func readYOrN(cmd *cobra.Command) bool {
 	var yOrN string
 	_, err := fmt.Scan(&yOrN)

cmd/config-converter/main.go

@@ -58,7 +58,7 @@ func main() {
 	}
 }
 
-func run(cmd *cobra.Command, args []string) {
+func run(_ *cobra.Command, _ []string) {
 	if cfgPath != "" && modelPath != "" {
 		fmt.Fprintln(os.Stderr, "can't specify both config and model")
 		os.Exit(ExitCodeInvalidFlag)

coordinator/controller.go

@@ -174,7 +174,7 @@ func NewController(
 	// detect the capture changes
 	c.nodeManager.RegisterNodeChangeHandler(
 		nodeChangeHandlerID,
-		func(allNodes map[node.ID]*node.Info) {
+		func(_ map[node.ID]*node.Info) {
 			c.nodeChanged.Lock()
 			defer c.nodeChanged.Unlock()
 			c.nodeChanged.changed = true
@@ -686,7 +686,7 @@ func (c *Controller) CreateChangefeed(ctx context.Context, info *config.ChangeFe
 			return errors.Trace(ctx.Err())
 		case <-ticker.C:
 			log.Warn("changefeed is in scheduling, wait a moment", zap.String("changefeed", info.ChangefeedID.DisplayName.String()))
-			count += 1
+			count++
 		}
 	}
 
@@ -728,7 +728,7 @@ func (c *Controller) RemoveChangefeed(ctx context.Context, id common.ChangeFeedI
 		case <-ctx.Done():
 			return 0, errors.Trace(ctx.Err())
 		case <-ticker.C:
-			count += 1
+			count++
 			log.Info("wait for stop changefeed operator finished", zap.Int("count", count), zap.Any("id", id))
 		}
 	}
@@ -766,7 +766,7 @@ func (c *Controller) PauseChangefeed(ctx context.Context, id common.ChangeFeedID
 		case <-ctx.Done():
 			return errors.Trace(ctx.Err())
 		case <-ticker.C:
-			count += 1
+			count++
 			log.Info("wait for stop changefeed operator finished", zap.Int("count", count), zap.Any("id", id))
 		}
 	}

downstreamadapter/eventcollector/event_collector.go

@@ -770,14 +770,6 @@ func updateMinUint64MapValue(m map[common.ChangeFeedID]uint64, key common.Change
 	}
 }
 
-func updateMaxUint64MapValue(m map[common.ChangeFeedID]uint64, key common.ChangeFeedID, value uint64) {
-	if existing, exists := m[key]; exists {
-		m[key] = max(existing, value)
-	} else {
-		m[key] = value
-	}
-}
-
 func calcUsageRatio(usedMemory int64, maxMemory int64) float64 {
 	if maxMemory <= 0 {
 		return 0

downstreamadapter/sink/mysql/causality/node.go

@@ -253,27 +253,6 @@ func (n *Node) getOrCreateDependers() *btree.BTreeG[*Node] {
 	return n.dependers
 }
 
-// dependerCount returns the number of dependers the node has.
-// NOTE: dependerCount is used for unit tests only.
-func (n *Node) dependerCount() int {
-	n.mu.Lock()
-	defer n.mu.Unlock()
-
-	if n.dependers == nil {
-		return 0
-	}
-	return n.dependers.Len()
-}
-
-// assignedWorkerID returns the cache ID that the node has been assigned to.
-// NOTE: assignedWorkerID is used for unit tests only.
-func (n *Node) assignedWorkerID() cacheID {
-	n.mu.Lock()
-	defer n.mu.Unlock()
-
-	return n.assignedTo
-}
-
 func genNextNodeID() int64 {
 	return nextNodeID.Add(1)
 }

downstreamadapter/sink/pulsar/helper.go

@@ -106,7 +106,7 @@ func newPulsarSinkComponentWithFactory(ctx context.Context,
 		return pulsarComponent, protocol, errors.Trace(err)
 	}
 
-	pulsarComponent.topicManager, err = topicmanager.GetPulsarTopicManagerAndTryCreateTopic(ctx, pulsarComponent.config, topic, pulsarComponent.client)
+	pulsarComponent.topicManager, err = topicmanager.GetPulsarTopicManagerAndTryCreateTopic(ctx, pulsarComponent.config, pulsarComponent.client)
 	if err != nil {
 		return pulsarComponent, protocol, errors.Trace(err)
 	}

downstreamadapter/sink/topicmanager/kafka_topic_manager.go

@@ -130,7 +130,7 @@ func (m *kafkaTopicManager) backgroundRefreshMeta(ctx context.Context) {
 		case <-m.metaRefreshTicker.C:
 			// We ignore the error here, because the error may be caused by the
 			// network problem, and we can try to get the metadata next time.
-			topicPartitionNums, _ := m.fetchAllTopicsPartitionsNum(ctx)
+			topicPartitionNums, _ := m.fetchAllTopicsPartitionsNum()
 			for topic, partitionNum := range topicPartitionNums {
 				m.tryUpdatePartitionsAndLogging(topic, partitionNum)
 			}
@@ -169,11 +169,9 @@ func (m *kafkaTopicManager) tryUpdatePartitionsAndLogging(topic string, partitio
 // The error returned by this method could be a transient error that is fixable by the underlying logic.
 // When handling this error, please be cautious.
 // If you simply throw the error to the caller, it may impact the robustness of your program.
-func (m *kafkaTopicManager) fetchAllTopicsPartitionsNum(
-	ctx context.Context,
-) (map[string]int32, error) {
+func (m *kafkaTopicManager) fetchAllTopicsPartitionsNum() (map[string]int32, error) {
 	var topics []string
-	m.topics.Range(func(key, value any) bool {
+	m.topics.Range(func(key, _ any) bool {
 		topics = append(topics, key.(string))
 		return true
 	})
@@ -244,7 +242,7 @@ func (m *kafkaTopicManager) waitUntilTopicVisible(
 // createTopic creates a topic with the given name
 // and returns the number of partitions.
 func (m *kafkaTopicManager) createTopic(
-	ctx context.Context,
+	_ context.Context,
 	topicName string,
 ) (int32, error) {
 	if !m.cfg.AutoCreate {

downstreamadapter/sink/topicmanager/pulsar_topic_manager.go

@@ -32,7 +32,6 @@ type pulsarTopicManager struct {
 func GetPulsarTopicManagerAndTryCreateTopic(
 	ctx context.Context,
 	cfg *config.PulsarConfig,
-	topic string,
 	client pulsar.Client,
 ) (TopicManager, error) {
 	topicManager := newPulsarTopicManager(cfg, client)

downstreamadapter/sink/topicmanager/pulsar_topic_manager_mock.go

@@ -15,9 +15,6 @@ package topicmanager
 
 import (
 	"context"
-
-	"github.com/apache/pulsar-client-go/pulsar"
-	"github.com/pingcap/ticdc/pkg/config"
 )
 
 // pulsarTopicManager is a manager for pulsar topics.
@@ -26,16 +23,13 @@ type pulsarTopicManagerMock struct {
 }
 
 // NewMockPulsarTopicManager creates a new topic manager.
-func NewMockPulsarTopicManager(
-	cfg *config.PulsarConfig,
-	client pulsar.Client,
-) (TopicManager, error) {
+func NewMockPulsarTopicManager() (TopicManager, error) {
 	mgr := &pulsarTopicManagerMock{}
 	return mgr, nil
 }
 
 // GetPartitionNum spend more time,but no use.
 // mock 3 partitions
-func (m *pulsarTopicManagerMock) GetPartitionNum(ctx context.Context, topic string) (int32, error) {
+func (m *pulsarTopicManagerMock) GetPartitionNum(_ context.Context, _ string) (int32, error) {
 	return 3, nil
 }

downstreamadapter/sink/topicmanager/pulsar_topic_manager_test.go

@@ -42,8 +42,6 @@ func newPulsarConfig(t *testing.T) (*config.PulsarConfig, *url.URL) {
 func TestGetPartitionNumMock(t *testing.T) {
 	t.Parallel()
 
-	cfg, _ := newPulsarConfig(t)
-
 	replicaConfig := config.GetDefaultReplicaConfig()
 	replicaConfig.Sink = &config.SinkConfig{
 		Protocol: util.AddressOf("canal-json"),
@@ -52,7 +50,7 @@ func TestGetPartitionNumMock(t *testing.T) {
 	ctx := context.Background()
 
 	ctx = context.WithValue(ctx, "testing.T", t)
-	pm, err := NewMockPulsarTopicManager(cfg, nil)
+	pm, err := NewMockPulsarTopicManager()
 	require.NoError(t, err)
 	require.NotNil(t, pm)
 

downstreamadapter/syncpoint/sync_point.go

@@ -31,9 +31,9 @@ func CalculateStartSyncPointTs(startTs uint64, syncPointInterval time.Duration,
 	}
 	k := int64(oracle.GetTimeFromTS(startTs).Sub(time.Unix(0, 0)) / syncPointInterval)
 	if oracle.GetTimeFromTS(startTs).Sub(time.Unix(0, 0))%syncPointInterval != 0 || oracle.ExtractLogical(startTs) != 0 {
-		k += 1
+		k++
 	} else if skipSyncpointAtStartTs {
-		k += 1
+		k++
 	}
 	return oracle.GoTimeToTS(time.Unix(0, 0).Add(time.Duration(int64(syncPointInterval) * k)))
 }

logservice/logpuller/errors.go

@@ -16,13 +16,10 @@ package logpuller
 import (
 	"fmt"
 
-	"github.com/pingcap/errors"
 	"github.com/pingcap/kvproto/pkg/cdcpb"
 	"github.com/tikv/client-go/v2/tikv"
 )
 
-var errUnreachable = errors.New("kv client unreachable error")
-
 type eventError struct {
 	err *cdcpb.Error
 }

logservice/logpuller/region_event_handler.go

@@ -169,7 +169,7 @@ func (h *regionEventHandler) GetSize(event regionEvent) int {
 	return event.getSize()
 }
 
-func (h *regionEventHandler) GetArea(path SubscriptionID, dest *subscribedSpan) int {
+func (h *regionEventHandler) GetArea(_ SubscriptionID, _ *subscribedSpan) int {
 	return 0
 }
 
@@ -198,7 +198,7 @@ func (h *regionEventHandler) GetTimestamp(event regionEvent) dynstream.Timestamp
 		return dynstream.Timestamp(event.resolvedTs)
 	}
 }
-func (h *regionEventHandler) IsPaused(event regionEvent) bool { return false }
+func (h *regionEventHandler) IsPaused(_ regionEvent) bool { return false }
 
 func (h *regionEventHandler) GetType(event regionEvent) dynstream.EventType {
 	if event.entries != nil || event.resolvedTs != 0 {

logservice/logpuller/subscription_client.go

@@ -319,7 +319,7 @@ func (s *subscriptionClient) updateMetrics(ctx context.Context) error {
 			}
 
 			pendingRegionReqCount := 0
-			s.stores.Range(func(key, value any) bool {
+			s.stores.Range(func(_, value any) bool {
 				store := value.(*requestedStore)
 				store.requestWorkers.RLock()
 				for _, worker := range store.requestWorkers.s {
@@ -581,7 +581,7 @@ func (s *subscriptionClient) handleRegions(ctx context.Context, eg *errgroup.Gro
 	}
 
 	defer func() {
-		s.stores.Range(func(key, value any) bool {
+		s.stores.Range(func(_, value any) bool {
 			rs := value.(*requestedStore)
 
 			rs.requestWorkers.RLock()
@@ -855,6 +855,7 @@ func (s *subscriptionClient) doHandleError(ctx context.Context, errInfo regionEr
 			zap.Error(err))
 	}
 
+	//nolint:errorlint // converting large type switch to errors.As is a significant refactor
 	switch eerr := err.(type) {
 	case *eventError:
 		innerErr := eerr.err
@@ -1046,7 +1047,7 @@ func (s *subscriptionClient) logSlowRegions(ctx context.Context) error {
 						zap.Any("slowRegion", attr.SlowestRegion))
 				}
 			} else if currTime.Sub(attr.SlowestRegion.Created) > 10*time.Minute {
-				slowInitializeRegion += 1
+				slowInitializeRegion++
 				log.Info("subscription client initializes a region too slow",
 					zap.Uint64("subscriptionID", uint64(subscriptionID)),
 					zap.Any("slowRegion", attr.SlowestRegion))

maintainer/barrier_event.go

@@ -742,8 +742,3 @@ func getAllNodes(nodeManager *watcher.NodeManager) []node.ID {
 	}
 	return nodes
 }
-
-// for test
-func (be *BarrierEvent) setLastResendTime(time time.Time) {
-	be.lastResendTime = time
-}

maintainer/replica/default_span_split_checker.go

@@ -29,10 +29,7 @@ import (
 	"go.uber.org/zap"
 )
 
-var (
-	trafficScoreThreshold = 3
-	regionScoreThreshold  = 3
-)
+var trafficScoreThreshold = 3
 
 // defaultSpanSplitChecker is used to check whether spans in the default group need to be split
 // based on multiple thresholds including write traffic and region count.