sink: fix max-batch-size (#1250) #1253

pingcap · Dec 30, 2020 · e65ce07 · e65ce07
1 parent 6b7fb14
commit e65ce07
Showing 1 changed file with 2 additions and 2 deletions.
diff --git a/cdc/sink/mq.go b/cdc/sink/mq.go
@@ -428,7 +428,7 @@ func newKafkaSaramaSink(ctx context.Context, sinkURI *url.URL, filter *filter.Fi
 
 	s = sinkURI.Query().Get("max-batch-size")
 	if s != "" {
-		opts["max-message-bytes"] = s
+		opts["max-batch-size"] = s
 	}
 
 	s = sinkURI.Query().Get("compression")
@@ -498,7 +498,7 @@ func newPulsarSink(ctx context.Context, sinkURI *url.URL, filter *filter.Filter,
 
 	s = sinkURI.Query().Get("max-batch-size")
 	if s != "" {
-		opts["max-message-bytes"] = s
+		opts["max-batch-size"] = s
 	}
 	// For now, it's a place holder. Avro format have to make connection to Schema Registery,
 	// and it may needs credential.