sink(ticdc): limit encoder-concurrency to avoid crash (pingcap#11775)

close pingcap#11773
mjonss · Nov 25, 2024 · 205b582 · 205b582
1 parent 94082c3
commit 205b582
Showing 1 changed file with 6 additions and 0 deletions.
diff --git a/pkg/sink/codec/encoder_group.go b/pkg/sink/codec/encoder_group.go
@@ -20,6 +20,7 @@ import (
 
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
+	"github.com/pingcap/tidb/pkg/util/cpu"
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/cdc/sink/dmlsink"
 	"github.com/pingcap/tiflow/pkg/config"
@@ -69,6 +70,11 @@ func NewEncoderGroup(
 	if concurrency <= 0 {
 		concurrency = config.DefaultEncoderGroupConcurrency
 	}
+	limitConcurrency := cpu.GetCPUCount() * 10
+	if concurrency > limitConcurrency {
+		concurrency = limitConcurrency
+		log.Warn("limit concurrency to avoid crash", zap.Int("concurrency", concurrency), zap.Any("limitConcurrency", limitConcurrency))
+	}
 	inputCh := make([]chan *future, concurrency)
 	for i := 0; i < concurrency; i++ {
 		inputCh[i] = make(chan *future, defaultInputChanSize)