Merge remote-tracking branch 'hyy/0905-2' into impl-puller-prior-queue

Author: asddongmen

logservice/logpuller/region_event_handler.go

@@ -128,20 +128,21 @@ func (h *regionEventHandler) GetArea(path SubscriptionID, dest *subscribedSpan)
 }
 
 func (h *regionEventHandler) GetTimestamp(event regionEvent) dynstream.Timestamp {
-	if event.entries != nil && event.entries.Entries != nil && len(event.entries.Entries.GetEntries()) > 0 {
-		entries := event.entries.Entries.GetEntries()
-		switch entries[0].Type {
-		case cdcpb.Event_INITIALIZED:
-			return dynstream.Timestamp(event.state.region.resolvedTs())
-		case cdcpb.Event_COMMITTED,
-			cdcpb.Event_PREWRITE,
-			cdcpb.Event_COMMIT,
-			cdcpb.Event_ROLLBACK:
-			return dynstream.Timestamp(entries[0].CommitTs)
-		default:
-			log.Warn("unknown event entries", zap.Any("event", event.entries))
-			return 0
+	if event.entries != nil && event.entries.Entries != nil {
+		for _, entry := range event.entries.Entries.GetEntries() {
+			switch entry.Type {
+			case cdcpb.Event_INITIALIZED:
+				return dynstream.Timestamp(event.state.region.resolvedTs())
+			case cdcpb.Event_COMMITTED,
+				cdcpb.Event_PREWRITE,
+				cdcpb.Event_COMMIT,
+				cdcpb.Event_ROLLBACK:
+				return dynstream.Timestamp(entry.CommitTs)
+			default:
+				// ignore other event types
+			}
 		}
+		return 0
 	} else {
 		return dynstream.Timestamp(event.resolvedTs)
 	}

logservice/logpuller/subscription_client.go

@@ -749,6 +749,7 @@ func (s *subscriptionClient) handleErrors(ctx context.Context) error {
 	for {
 		select {
 		case <-ctx.Done():
+			log.Info("subscription client handle errors exit")
 			return ctx.Err()
 		case errInfo := <-s.errCache.errCh:
 			if err := s.doHandleError(ctx, errInfo); err != nil {
@@ -1086,7 +1087,11 @@ func (e *errCache) dispatch(ctx context.Context) error {
 		errInfo := e.cache[0]
 		e.cache = e.cache[1:]
 		e.Unlock()
-		e.errCh <- errInfo
+		select {
+		case <-ctx.Done():
+			log.Info("subscription client dispatch err cache done")
+		case e.errCh <- errInfo:
+		}
 	}
 	for {
 		select {

logservice/logpuller/subscription_client_test.go

@@ -204,3 +204,60 @@ func TestSubscriptionWithFailedTiKV(t *testing.T) {
 		require.True(t, false, "reconnection not succeed in 5 second")
 	}
 }
+
+// TestErrCacheDispatchWithFullChannelAndCanceledContext tests that when errCh is full
+// and context is canceled, the dispatch method doesn't get stuck.
+func TestErrCacheDispatchWithFullChannelAndCanceledContext(t *testing.T) {
+	// Create errCache with a small errCh to easily fill it up
+	errCache := &errCache{
+		cache:  make([]regionErrorInfo, 0, 10),
+		errCh:  make(chan regionErrorInfo, 2), // Small buffer to easily fill
+		notify: make(chan struct{}, 10),
+	}
+
+	// Create a mock regionErrorInfo
+	mockErrInfo := regionErrorInfo{
+		regionInfo: regionInfo{
+			verID: tikv.NewRegionVerID(1, 1, 1),
+			span:  heartbeatpb.TableSpan{TableID: 1, StartKey: []byte("a"), EndKey: []byte("b")},
+		},
+		err: errors.New("test error"),
+	}
+
+	// Fill up the errCh channel to make it full
+	errCache.errCh <- mockErrInfo
+	errCache.errCh <- mockErrInfo
+
+	// Add some errors to the cache
+	for i := 0; i < 5; i++ {
+		errCache.add(mockErrInfo)
+	}
+
+	// Create a context that will be canceled
+	ctx, cancel := context.WithCancel(context.Background())
+
+	// Channel to signal when dispatch returns
+	dispatchDone := make(chan error, 1)
+
+	// Start dispatch in a goroutine
+	go func() {
+		err := errCache.dispatch(ctx)
+		dispatchDone <- err
+	}()
+
+	// Give dispatch some time to start and potentially get stuck
+	time.Sleep(50 * time.Millisecond)
+
+	// Cancel the context
+	cancel()
+
+	// Wait for dispatch to return with a timeout
+	select {
+	case err := <-dispatchDone:
+		// Verify that dispatch returned with context.Canceled error
+		require.Equal(t, context.Canceled, err)
+	case <-time.After(5 * time.Second):
+		// If we timeout here, it means dispatch is stuck
+		t.Fatal("dispatch method is stuck and didn't return after context cancellation")
+	}
+}

maintainer/operator/operator_controller.go

@@ -203,8 +203,8 @@ func (oc *Controller) GetMinCheckpointTs() uint64 {
 		if op.BlockTsForward() {
 			spanReplication := oc.spanController.GetTaskByID(op.ID())
 			if spanReplication == nil {
-				// for test
-				log.Panic("span replication is nil", zap.String("operator", op.String()))
+				log.Info("span replication is nil", zap.String("operator", op.String()))
+				continue
 			}
 			if spanReplication.GetStatus().CheckpointTs < minCheckpointTs {
 				minCheckpointTs = spanReplication.GetStatus().CheckpointTs

maintainer/replica/default_span_split_checker.go

@@ -128,7 +128,7 @@ func (s *defaultSpanSplitChecker) UpdateStatus(replica *SpanReplication) {
 
 	// check traffic first
 	if status.GetStatus().EventSizePerSecond != 0 {
-		if status.GetStatus().EventSizePerSecond < float32(s.writeThreshold) {
+		if s.writeThreshold == 0 || status.GetStatus().EventSizePerSecond < float32(s.writeThreshold) {
 			status.trafficScore = 0
 		} else {
 			status.trafficScore++
@@ -147,7 +147,7 @@ func (s *defaultSpanSplitChecker) UpdateStatus(replica *SpanReplication) {
 		status.regionCheckTime = time.Now()
 	}
 
-	log.Info("default span split checker: update status", zap.String("changefeed", s.changefeedID.Name()), zap.String("replica", replica.ID.String()), zap.Int("trafficScore", status.trafficScore), zap.Int("regionCount", status.regionCount))
+	log.Debug("default span split checker: update status", zap.String("changefeed", s.changefeedID.Name()), zap.String("replica", replica.ID.String()), zap.Int("trafficScore", status.trafficScore), zap.Int("regionCount", status.regionCount))
 
 	if status.trafficScore >= trafficScoreThreshold || (s.regionThreshold > 0 && status.regionCount >= s.regionThreshold) {
 		if _, ok := s.splitReadyTasks[status.ID]; !ok {

server/module_http.go

@@ -22,8 +22,6 @@ import (
 	"github.com/gin-gonic/gin"
 	"github.com/pingcap/log"
 	"github.com/pingcap/ticdc/api"
-	"github.com/pingcap/ticdc/pkg/common"
-	"github.com/pingcap/ticdc/pkg/errors"
 	"github.com/pingcap/ticdc/pkg/logger"
 	"go.uber.org/zap"
 	"golang.org/x/net/netutil"
@@ -43,7 +41,7 @@ type HttpServer struct {
 
 // NewHttpServer create the HTTP server.
 // `lis` is a listener that gives us plain-text HTTP requests.
-func NewHttpServer(c *server, lis net.Listener) common.SubModule {
+func NewHttpServer(c *server, lis net.Listener) *HttpServer {
 	// LimitListener returns a Listener that accepts at most n simultaneous
 	// connections from the provided Listener. Connections that exceed the
 	// limit will wait in a queue and no new goroutines will be created until
@@ -72,17 +70,24 @@ func NewHttpServer(c *server, lis net.Listener) common.SubModule {
 	}
 }
 
-func (s *HttpServer) Run(ctx context.Context) error {
+func (s *HttpServer) Run(ctx context.Context) {
 	log.Info("http server is running", zap.String("addr", s.listener.Addr().String()))
-	err := s.server.Serve(s.listener)
-	if err != nil {
-		log.Error("http server error", zap.Error(err))
-	}
-	return errors.WrapError(errors.ErrServeHTTP, err)
+	go func() {
+		err := s.server.Serve(s.listener)
+		if err != nil {
+			log.Error("http server error", zap.Error(err))
+		}
+	}()
 }
 
-func (s *HttpServer) Close(ctx context.Context) error {
-	return s.server.Shutdown(ctx)
+func (s *HttpServer) Close() {
+	log.Info("http server is closing")
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	err := s.server.Shutdown(ctx)
+	if err != nil {
+		log.Warn("close http server failed", zap.Error(err))
+	}
 }
 
 func (s *HttpServer) Name() string {

server/server.go

@@ -173,7 +173,6 @@ func (c *server) initialize(ctx context.Context) error {
 	c.subCommonModules = []common.SubModule{
 		nodeManager,
 		NewElector(c),
-		NewHttpServer(c, c.tcpServer.HTTP1Listener()),
 		NewGrpcServer(c.tcpServer.GrpcListener()),
 	}
 
@@ -232,6 +231,10 @@ func (c *server) setPreServices(ctx context.Context) error {
 	appctx.SetService(appctx.DispatcherOrchestrator, dispatcherOrchestrator)
 	c.preServices = append(c.preServices, dispatcherOrchestrator)
 
+	httpServer := NewHttpServer(c, c.tcpServer.HTTP1Listener())
+	httpServer.Run(ctx)
+	c.preServices = append(c.preServices, httpServer)
+
 	log.Info("pre services all set", zap.Any("preServicesNum", len(c.preServices)))
 	return nil
 }