2 files changed, 28 insertions, 13 deletions
diff --git a/weed/mq/client/cmd/weed_pub/publisher.go b/weed/mq/client/cmd/weed_pub/publisher.go
index ee00be9f8..59469e66b 100644
--- a/weed/mq/client/cmd/weed_pub/publisher.go
+++ b/weed/mq/client/cmd/weed_pub/publisher.go
@@ -12,7 +12,8 @@ import (
 
 var (
 	messageCount = flag.Int("n", 1000, "message count")
-	concurrency  = flag.Int("c", 4, "concurrency count")
+	concurrency  = flag.Int("c", 4, "concurrent publishers")
+	partitionCount = flag.Int("p", 6, "partition count")
 
 	namespace   = flag.String("ns", "test", "namespace")
 	topic       = flag.String("topic", "test", "topic")
@@ -25,7 +26,10 @@ func doPublish(publisher *pub_client.TopicPublisher, id int) {
 		// Simulate publishing a message
 		key := []byte(fmt.Sprintf("key-%d-%d", id, i))
 		value := []byte(fmt.Sprintf("value-%d-%d", id, i))
-		publisher.Publish(key, value) // Call your publisher function here
+		if err := publisher.Publish(key, value); err != nil {
+			fmt.Println(err)
+			break
+		}
 		// println("Published", string(key), string(value))
 	}
 	elapsed := time.Since(startTime)
@@ -35,7 +39,8 @@ func doPublish(publisher *pub_client.TopicPublisher, id int) {
 func main() {
 	flag.Parse()
 	config := &pub_client.PublisherConfiguration{
-		CreateTopic: true,
+		CreateTopic:               true,
+		CreateTopicPartitionCount: int32(*partitionCount),
 	}
 	publisher := pub_client.NewTopicPublisher(*namespace, *topic, config)
 	brokers := strings.Split(*seedBrokers, ",")
diff --git a/weed/mq/client/cmd/weed_sub/subscriber.go b/weed/mq/client/cmd/weed_sub/subscriber.go
index d5bd8f12d..a175c948a 100644
--- a/weed/mq/client/cmd/weed_sub/subscriber.go
+++ b/weed/mq/client/cmd/weed_sub/subscriber.go
@@ -3,7 +3,9 @@ package main
 import (
 	"flag"
 	"fmt"
+	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/mq/client/sub_client"
+	"github.com/seaweedfs/seaweedfs/weed/util"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
 	"strings"
@@ -14,35 +16,43 @@ var (
 	namespace   = flag.String("ns", "test", "namespace")
 	topic       = flag.String("topic", "test", "topic")
 	seedBrokers = flag.String("brokers", "localhost:17777", "seed brokers")
+
+	clientId = flag.Uint("client_id", uint(util.RandomInt32()), "client id")
 )
 
 func main() {
 	flag.Parse()
 
 	subscriberConfig := &sub_client.SubscriberConfiguration{
-		ClientId:        "testSubscriber",
-		GroupId:         "test",
-		GroupInstanceId: "test",
-		GrpcDialOption:  grpc.WithTransportCredentials(insecure.NewCredentials()),
+		ClientId:                fmt.Sprintf("client-%d", *clientId),
+		ConsumerGroup:           "test",
+		ConsumerGroupInstanceId: fmt.Sprintf("client-%d", *clientId),
+		GrpcDialOption:          grpc.WithTransportCredentials(insecure.NewCredentials()),
 	}
 
 	contentConfig := &sub_client.ContentConfiguration{
 		Namespace: *namespace,
 		Topic:     *topic,
 		Filter:    "",
-		StartTime: time.Now(),
+		StartTime: time.Unix(1, 1),
+	}
+
+	processorConfig := sub_client.ProcessorConfiguration{
+		ConcurrentPartitionLimit: 3,
 	}
 
 	brokers := strings.Split(*seedBrokers, ",")
-	subscriber := sub_client.NewTopicSubscriber(brokers, subscriberConfig, contentConfig)
+	subscriber := sub_client.NewTopicSubscriber(brokers, subscriberConfig, contentConfig, processorConfig)
 
-	subscriber.SetEachMessageFunc(func(key, value []byte) bool {
-		println(string(key), "=>", string(value))
-		return true
+	counter := 0
+	subscriber.SetEachMessageFunc(func(key, value []byte) (bool, error) {
+		counter++
+		println(string(key), "=>", string(value), counter)
+		return true, nil
 	})
 
 	subscriber.SetCompletionFunc(func() {
-		println("done subscribing")
+		glog.V(0).Infof("done recived %d messages", counter)
 	})
 
 	if err := subscriber.Subscribe(); err != nil {