feat(kq): support custom producer balancer and message key

Author: testerxiaodong

kq/pusher.go

@@ -22,6 +22,7 @@ type (
 	pushOptions struct {
 		// kafka.Writer options
 		allowAutoTopicCreation bool
+		balancer               kafka.Balancer
 
 		// executors.ChunkExecutor options
 		chunkSize     int
@@ -45,6 +46,9 @@ func NewPusher(addrs []string, topic string, opts ...PushOption) *Pusher {
 
 	// apply kafka.Writer options
 	producer.AllowAutoTopicCreation = options.allowAutoTopicCreation
+	if options.balancer != nil {
+		producer.Balancer = options.balancer
+	}
 
 	// apply ChunkExecutor options
 	var chunkOpts []executors.ChunkOption
@@ -139,3 +143,10 @@ func WithAllowAutoTopicCreation() PushOption {
 		options.allowAutoTopicCreation = true
 	}
 }
+
+// WithBalancer customizes the Pusher with the given balancer.
+func WithBalancer(balancer kafka.Balancer) PushOption {
+	return func(options *pushOptions) {
+		options.balancer = balancer
+	}
+}