idempotent = true

Signed-off-by: Ping Yu <[email protected]>
tikv · Mar 3, 2024 · 1a15bc5 · 1a15bc5
1 parent acf0d83
commit 1a15bc5
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 1 deletion.
diff --git a/cdc/cdc/sink/producer/kafka/config.go b/cdc/cdc/sink/producer/kafka/config.go
@@ -231,6 +231,7 @@ func newSaramaConfig(ctx context.Context, c *Config) (*sarama.Config, error) {
 	// and https://github.com/tikv/migration/cdc/issues/3352.
 	config.Metadata.Timeout = 1 * time.Minute
 
+	config.Producer.Idempotent = true
 	config.Producer.Partitioner = sarama.NewManualPartitioner
 	config.Producer.MaxMessageBytes = c.MaxMessageBytes
 	config.Producer.Return.Successes = true
@@ -260,6 +261,8 @@ func newSaramaConfig(ctx context.Context, c *Config) (*sarama.Config, error) {
 	config.Admin.Retry.Backoff = 500 * time.Millisecond
 	config.Admin.Timeout = 1 * time.Minute
 
+	config.Net.MaxOpenRequests = 1
+
 	if c.Credential != nil && len(c.Credential.CAPath) != 0 {
 		config.Net.TLS.Enable = true
 		config.Net.TLS.Config, err = c.Credential.ToTLSConfig()

diff --git a/cdc/tests/integration_tests/changefeed_pause_resume/run.sh b/cdc/tests/integration_tests/changefeed_pause_resume/run.sh
@@ -28,7 +28,7 @@ function run() {
 		run_kafka_consumer --workdir "$WORK_DIR" --upstream-uri "$SINK_URI"
 	fi
 
-	for _ in $(seq 1 10); do
+	for _ in $(seq 1 100000); do
 		tikv-cdc cli changefeed pause --changefeed-id=$changefeed_id --pd=$UP_PD
 		rawkv_op $UP_PD put 5000
 		tikv-cdc cli changefeed resume --changefeed-id=$changefeed_id --pd=$UP_PD