[close tikv#381] Add integration tests for Kafka sink (tikv#382)

* wip Signed-off-by: Ping Yu <[email protected]> * add go:generate Signed-off-by: Ping Yu <[email protected]> * add tools/Makefile Signed-off-by: Ping Yu <[email protected]> * fix build error Signed-off-by: Ping Yu <[email protected]> * built Signed-off-by: Ping Yu <[email protected]> * fix Signed-off-by: Ping Yu <[email protected]> * fix ut Signed-off-by: Ping Yu <[email protected]> * clean up Signed-off-by: Ping Yu <[email protected]> * fix ut Signed-off-by: Ping Yu <[email protected]> * add ut Signed-off-by: Ping Yu <[email protected]> * fix check Signed-off-by: Ping Yu <[email protected]> * first works Signed-off-by: Ping Yu <[email protected]> * kafka sink in run.sh Signed-off-by: Ping Yu <[email protected]> * fix failpoint Signed-off-by: Ping Yu <[email protected]> * fix fmt Signed-off-by: Ping Yu <[email protected]> * fix sigstop Signed-off-by: Ping Yu <[email protected]> * fix check Signed-off-by: Ping Yu <[email protected]> * skip not support case Signed-off-by: Ping Yu <[email protected]> * default to kafka Signed-off-by: Ping Yu <[email protected]> * update readme.md in tests Signed-off-by: Ping Yu <[email protected]> * polish Signed-off-by: Ping Yu <[email protected]> --------- Signed-off-by: Ping Yu <[email protected]>
pingyu · Jan 8, 2024 · aeae68f · aeae68f
1 parent 99b7b74
commit aeae68f
Show file tree

Hide file tree

Showing 47 changed files with 965 additions and 43 deletions.
diff --git a/cdc/Makefile b/cdc/Makefile
@@ -73,6 +73,8 @@ LDFLAGS += -X "$(CDC_PKG)/pkg/version.GoVersion=$(GOVERSION)"
 
 SCVERSION := stable
 
+TEST_ON_BRANCH ?= release-6.5
+
 include tools/Makefile
 
 default: build buildsucc
@@ -208,7 +210,7 @@ rawkv_data:
 cdc_state_checker:
 	$(GOBUILD) -ldflags '$(LDFLAGS)' -o bin/cdc_state_checker ./tests/utils/cdc_state_checker/
 
-integration_test_build: check_failpoint_ctl rawkv_data cdc_state_checker
+integration_test_build: check_failpoint_ctl rawkv_data cdc_state_checker kafka_consumer
 	$(FAILPOINT_ENABLE)
 	$(GOTEST) -ldflags '$(LDFLAGS)' -c -cover -covermode=atomic \
         -coverpkg=github.com/tikv/migration/... \
@@ -219,8 +221,12 @@ integration_test_build: check_failpoint_ctl rawkv_data cdc_state_checker
 	$(FAILPOINT_DISABLE)
 
 integration_test: prepare_test_binaries check_third_party_binary integration_test_build
-	tests/integration_tests/run.sh "$(CASE)"
-	@bash <(curl -s https://codecov.io/bash) -F cdc -s /tmp/tikv_cdc_test -f *.out -t $(TIKV_MIGRATION_CODECOV_TOKEN)
+	tests/integration_tests/run.sh tikv "$(CASE)"
+	@bash <(curl -s https://codecov.io/bash) -F cdc -s /tmp/tikv_cdc_test -f *.out -t "$(TIKV_MIGRATION_CODECOV_TOKEN)"
+
+integration_test_kafka: prepare_test_binaries check_third_party_binary integration_test_build
+	tests/integration_tests/run.sh kafka "$(CASE)"
+	@bash <(curl -s https://codecov.io/bash) -F cdc -s /tmp/tikv_cdc_test -f *.out -t "$(TIKV_MIGRATION_CODECOV_TOKEN)"
 
 # To make try on integration tests easier.
 # Comment out the succeed cases and retry from the failed one.
@@ -242,7 +248,7 @@ integration_test_by_group: prepare_test_binaries check_third_party_binary integr
 	tests/integration_tests/run_group.sh others
 
 prepare_test_binaries:
-	cd scripts && ./download-integration-test-binaries.sh master && cd ..
+	cd scripts && ./download-integration-test-binaries.sh "$(TEST_ON_BRANCH)" && cd ..
 	touch prepare_test_binaries
 
 check_third_party_binary:

diff --git a/cdc/cdc/sink/codec/json.go b/cdc/cdc/sink/codec/json.go
@@ -505,11 +505,11 @@ func (b *JSONEventBatchDecoder) NextChangedEvent() (*model.RawKVEntry, error) {
 	valueLen := binary.BigEndian.Uint64(b.valueBytes[:8])
 	value := b.valueBytes[8 : valueLen+8]
 	b.valueBytes = b.valueBytes[valueLen+8:]
-	mvMsg := new(mqMessageValue)
-	if err := mvMsg.Decode(value); err != nil {
+	kvMsg := new(mqMessageValue)
+	if err := kvMsg.Decode(value); err != nil {
 		return nil, errors.Trace(err)
 	}
-	kvEvent := mqMessageToKvEvent(b.nextKey, mvMsg)
+	kvEvent := mqMessageToKvEvent(b.nextKey, kvMsg)
 	b.nextKey = nil
 	return kvEvent, nil
 }

diff --git a/cdc/cdc/sink/mq.go b/cdc/cdc/sink/mq.go
@@ -21,6 +21,7 @@ import (
 	"time"
 
 	"github.com/pingcap/errors"
+	"github.com/pingcap/failpoint"
 	"github.com/pingcap/log"
 	"github.com/tikv/migration/cdc/cdc/model"
 	"github.com/tikv/migration/cdc/cdc/sink/codec"
@@ -243,6 +244,11 @@ func (k *mqSink) runWorker(ctx context.Context, partition int32) error {
 				return 0, nil
 			}
 
+			failpoint.Inject("SinkFlushEventPanic", func() {
+				time.Sleep(time.Second)
+				log.Fatal("kafka sink injected error")
+			})
+
 			for _, msg := range messages {
 				err := k.writeToProducer(ctx, msg, codec.EncoderNeedAsyncWrite, partition)
 				if err != nil {