test: modify debezium integration test (#3537)

ref #442

Author: wk989898

tests/integration_tests/debezium01/src/test_cases.go

@@ -36,7 +36,7 @@ import (
 	"go.uber.org/zap"
 )
 
-const timeout = time.Second * 20
+const timeout = time.Second * 30
 
 var (
 	nFailed = 0
@@ -164,14 +164,16 @@ func runTestCase(testCasePath, dbConnMySQL, dbConnTiDB string) {
 	}
 }
 
-func fetchNextCDCRecord(reader *kafka.Reader, kind Kind, timeout time.Duration) (map[string]any, map[string]any, error) {
+func fetchNextCDCRecord(reader *kafka.Reader, kind Kind, timeout time.Duration, needWait bool, query string) (map[string]any, map[string]any, error) {
 	ctx, cancel := context.WithTimeout(context.Background(), timeout)
 	defer cancel()
 	for {
 		m, err := reader.FetchMessage(ctx)
 		if err != nil {
 			if errors.Is(err, context.DeadlineExceeded) {
-				logger.Warn("fetch record timed out", zap.Error(err))
+				if needWait {
+					logger.Warn("fetch record timed out", zap.String("query", query), zap.Any("kind", kind), zap.Error(err))
+				}
 				return nil, nil, nil
 			}
 			return nil, nil, fmt.Errorf("Failed to read CDC record of %s: %w", kind, err)
@@ -299,14 +301,14 @@ func runSingleQuery(query string, waitCDCRows bool) bool {
 		wg := &sync.WaitGroup{}
 		wg.Add(2)
 		go func() {
-			if _, _, err := fetchNextCDCRecord(readerDebezium, KindMySQL, timeout); err != nil {
-				logger.Error("fetch record failed", zap.Error(err))
+			if _, _, err := fetchNextCDCRecord(readerDebezium, KindMySQL, timeout, false, query); err != nil {
+				logger.Error("fetch record failed", zap.String("query", query), zap.Error(err))
 			}
 			wg.Done()
 		}()
 		go func() {
-			if _, _, err := fetchNextCDCRecord(readerTiCDC, KindTiDB, timeout); err != nil {
-				logger.Error("fetch record failed", zap.Error(err))
+			if _, _, err := fetchNextCDCRecord(readerTiCDC, KindTiDB, timeout, false, query); err != nil {
+				logger.Error("fetch record failed", zap.String("query", query), zap.Error(err))
 			}
 			wg.Done()
 		}()
@@ -333,17 +335,17 @@ func runSingleQuery(query string, waitCDCRows bool) bool {
 		wg.Add(2)
 		go func() {
 			var err error
-			keyMapsDebezium, objsDebezium, err = fetchNextCDCRecord(readerDebezium, KindMySQL, timeout)
+			keyMapsDebezium, objsDebezium, err = fetchNextCDCRecord(readerDebezium, KindMySQL, timeout, true, query)
 			if err != nil {
-				logger.Error("fetch record failed", zap.Error(err))
+				logger.Error("fetch record failed", zap.String("query", query), zap.Error(err))
 			}
 			wg.Done()
 		}()
 		go func() {
 			var err error
-			keyMapsTiCDC, objsTiCDC, err = fetchNextCDCRecord(readerTiCDC, KindTiDB, timeout)
+			keyMapsTiCDC, objsTiCDC, err = fetchNextCDCRecord(readerTiCDC, KindTiDB, timeout, true, query)
 			if err != nil {
-				logger.Error("fetch record failed", zap.Error(err))
+				logger.Error("fetch record failed", zap.String("query", query), zap.Error(err))
 			}
 			wg.Done()
 		}()

tests/integration_tests/run_heavy_it_in_ci.sh

@@ -96,7 +96,7 @@ kafka_groups=(
 	# G14
 	'debezium02'
 	# G15
-	# 'debezium03'
+	'debezium03'
 )
 
 # 12 CPU cores will be allocated to run each pulsar heavy group in CI pipelines.