aboutsummaryrefslogtreecommitdiff
path: root/weed/mq/client
diff options
context:
space:
mode:
Diffstat (limited to 'weed/mq/client')
-rw-r--r--weed/mq/client/cmd/weed_pub/publisher.go11
-rw-r--r--weed/mq/client/cmd/weed_sub/subscriber.go30
-rw-r--r--weed/mq/client/pub_client/connect.go11
-rw-r--r--weed/mq/client/pub_client/lookup.go4
-rw-r--r--weed/mq/client/pub_client/publish.go6
-rw-r--r--weed/mq/client/pub_client/publisher.go2
-rw-r--r--weed/mq/client/sub_client/connect_to_sub_coordinator.go171
-rw-r--r--weed/mq/client/sub_client/lookup.go34
-rw-r--r--weed/mq/client/sub_client/process.go85
-rw-r--r--weed/mq/client/sub_client/subscribe.go41
-rw-r--r--weed/mq/client/sub_client/subscriber.go18
11 files changed, 227 insertions, 186 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 {
diff --git a/weed/mq/client/pub_client/connect.go b/weed/mq/client/pub_client/connect.go
index fc7ff4d77..7f6d62a67 100644
--- a/weed/mq/client/pub_client/connect.go
+++ b/weed/mq/client/pub_client/connect.go
@@ -21,17 +21,17 @@ func (p *TopicPublisher) doConnect(partition *mq_pb.Partition, brokerAddress str
return publishClient, fmt.Errorf("dial broker %s: %v", brokerAddress, err)
}
brokerClient := mq_pb.NewSeaweedMessagingClient(grpcConnection)
- stream, err := brokerClient.Publish(context.Background())
+ stream, err := brokerClient.PublishMessage(context.Background())
if err != nil {
return publishClient, fmt.Errorf("create publish client: %v", err)
}
publishClient = &PublishClient{
- SeaweedMessaging_PublishClient: stream,
+ SeaweedMessaging_PublishMessageClient: stream,
Broker: brokerAddress,
}
- if err = publishClient.Send(&mq_pb.PublishRequest{
- Message: &mq_pb.PublishRequest_Init{
- Init: &mq_pb.PublishRequest_InitMessage{
+ if err = publishClient.Send(&mq_pb.PublishMessageRequest{
+ Message: &mq_pb.PublishMessageRequest_Init{
+ Init: &mq_pb.PublishMessageRequest_InitMessage{
Topic: &mq_pb.Topic{
Namespace: p.namespace,
Name: p.topic,
@@ -40,6 +40,7 @@ func (p *TopicPublisher) doConnect(partition *mq_pb.Partition, brokerAddress str
RingSize: partition.RingSize,
RangeStart: partition.RangeStart,
RangeStop: partition.RangeStop,
+ UnixTimeNs: partition.UnixTimeNs,
},
AckInterval: 128,
},
diff --git a/weed/mq/client/pub_client/lookup.go b/weed/mq/client/pub_client/lookup.go
index e55bfd256..ccc83b58d 100644
--- a/weed/mq/client/pub_client/lookup.go
+++ b/weed/mq/client/pub_client/lookup.go
@@ -3,6 +3,7 @@ package pub_client
import (
"context"
"fmt"
+ "github.com/seaweedfs/seaweedfs/weed/glog"
"github.com/seaweedfs/seaweedfs/weed/pb"
"github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
)
@@ -39,6 +40,7 @@ func (p *TopicPublisher) doLookupAndConnect(brokerAddress string) error {
},
IsForPublish: true,
})
+ glog.V(0).Infof("lookup1 topic %s/%s: %v", p.namespace, p.topic, lookupResp)
if p.config.CreateTopic && err != nil {
_, err = client.ConfigureTopic(context.Background(), &mq_pb.ConfigureTopicRequest{
Topic: &mq_pb.Topic{
@@ -58,12 +60,14 @@ func (p *TopicPublisher) doLookupAndConnect(brokerAddress string) error {
},
IsForPublish: true,
})
+ glog.V(0).Infof("lookup2 topic %s/%s: %v", p.namespace, p.topic, lookupResp)
}
if err != nil {
return err
}
for _, brokerPartitionAssignment := range lookupResp.BrokerPartitionAssignments {
+ glog.V(0).Infof("topic %s/%s partition %v leader %s followers %v", p.namespace, p.topic, brokerPartitionAssignment.Partition, brokerPartitionAssignment.LeaderBroker, brokerPartitionAssignment.FollowerBrokers)
// partition => publishClient
publishClient, err := p.doConnect(brokerPartitionAssignment.Partition, brokerPartitionAssignment.LeaderBroker)
if err != nil {
diff --git a/weed/mq/client/pub_client/publish.go b/weed/mq/client/pub_client/publish.go
index 1e250ede3..4b0dfade9 100644
--- a/weed/mq/client/pub_client/publish.go
+++ b/weed/mq/client/pub_client/publish.go
@@ -12,7 +12,7 @@ func (p *TopicPublisher) Publish(key, value []byte) error {
if hashKey < 0 {
hashKey = -hashKey
}
- publishClient, found := p.partition2Broker.Floor(hashKey, hashKey)
+ publishClient, found := p.partition2Broker.Floor(hashKey+1, hashKey+1)
if !found {
return fmt.Errorf("no broker found for key %d", hashKey)
}
@@ -27,8 +27,8 @@ func (p *TopicPublisher) Publish(key, value []byte) error {
//google.golang.org/grpc.(*clientStream).SendMsg(stream.go:894)
//github.com/seaweedfs/seaweedfs/weed/pb/mq_pb.(*seaweedMessagingPublishClient).Send(mq_grpc.pb.go:141)
//github.com/seaweedfs/seaweedfs/weed/mq/client/pub_client.(*TopicPublisher).Publish(publish.go:19)
- if err := publishClient.Send(&mq_pb.PublishRequest{
- Message: &mq_pb.PublishRequest_Data{
+ if err := publishClient.Send(&mq_pb.PublishMessageRequest{
+ Message: &mq_pb.PublishMessageRequest_Data{
Data: &mq_pb.DataMessage{
Key: key,
Value: value,
diff --git a/weed/mq/client/pub_client/publisher.go b/weed/mq/client/pub_client/publisher.go
index a0c26db36..d5176f21b 100644
--- a/weed/mq/client/pub_client/publisher.go
+++ b/weed/mq/client/pub_client/publisher.go
@@ -17,7 +17,7 @@ type PublisherConfiguration struct {
}
type PublishClient struct {
- mq_pb.SeaweedMessaging_PublishClient
+ mq_pb.SeaweedMessaging_PublishMessageClient
Broker string
Err error
}
diff --git a/weed/mq/client/sub_client/connect_to_sub_coordinator.go b/weed/mq/client/sub_client/connect_to_sub_coordinator.go
new file mode 100644
index 000000000..6d0eb0d43
--- /dev/null
+++ b/weed/mq/client/sub_client/connect_to_sub_coordinator.go
@@ -0,0 +1,171 @@
+package sub_client
+
+import (
+ "context"
+ "fmt"
+ "github.com/seaweedfs/seaweedfs/weed/glog"
+ "github.com/seaweedfs/seaweedfs/weed/pb"
+ "github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
+ "io"
+ "sync"
+ "time"
+)
+
+func (sub *TopicSubscriber) doKeepConnectedToSubCoordinator() {
+ waitTime := 1 * time.Second
+ for {
+ for _, broker := range sub.bootstrapBrokers {
+ // TODO find the balancer
+ // connect to the balancer
+ pb.WithBrokerGrpcClient(true, broker, sub.SubscriberConfig.GrpcDialOption, func(client mq_pb.SeaweedMessagingClient) error {
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+
+ stream, err := client.SubscriberToSubCoordinator(ctx)
+ if err != nil {
+ glog.V(0).Infof("subscriber %s/%s: %v", sub.ContentConfig.Namespace, sub.ContentConfig.Topic, err)
+ return err
+ }
+ waitTime = 1 * time.Second
+
+ // Maybe later: subscribe to multiple topics instead of just one
+
+ if err := stream.Send(&mq_pb.SubscriberToSubCoordinatorRequest{
+ Message: &mq_pb.SubscriberToSubCoordinatorRequest_Init{
+ Init: &mq_pb.SubscriberToSubCoordinatorRequest_InitMessage{
+ ConsumerGroup: sub.SubscriberConfig.ConsumerGroup,
+ ConsumerGroupInstanceId: sub.SubscriberConfig.ConsumerGroupInstanceId,
+ Topic: &mq_pb.Topic{
+ Namespace: sub.ContentConfig.Namespace,
+ Name: sub.ContentConfig.Topic,
+ },
+ },
+ },
+ }); err != nil {
+ glog.V(0).Infof("subscriber %s/%s send init: %v", sub.ContentConfig.Namespace, sub.ContentConfig.Topic, err)
+ return err
+ }
+
+ // keep receiving messages from the sub coordinator
+ for {
+ resp, err := stream.Recv()
+ if err != nil {
+ glog.V(0).Infof("subscriber %s/%s receive: %v", sub.ContentConfig.Namespace, sub.ContentConfig.Topic, err)
+ return err
+ }
+ assignment := resp.GetAssignment()
+ if assignment != nil {
+ glog.V(0).Infof("subscriber %s/%s receive assignment: %v", sub.ContentConfig.Namespace, sub.ContentConfig.Topic, assignment)
+ }
+ sub.onEachAssignment(assignment)
+ }
+
+ return nil
+ })
+ }
+ glog.V(0).Infof("subscriber %s/%s/%s waiting for more assignments", sub.ContentConfig.Namespace, sub.ContentConfig.Topic, sub.SubscriberConfig.ConsumerGroup)
+ if waitTime < 10*time.Second {
+ waitTime += 1 * time.Second
+ }
+ time.Sleep(waitTime)
+ }
+}
+
+func (sub *TopicSubscriber) onEachAssignment(assignment *mq_pb.SubscriberToSubCoordinatorResponse_Assignment) {
+ if assignment == nil {
+ return
+ }
+ // process each partition, with a concurrency limit
+ var wg sync.WaitGroup
+ semaphore := make(chan struct{}, sub.ProcessorConfig.ConcurrentPartitionLimit)
+
+ for _, assigned := range assignment.AssignedPartitions {
+ wg.Add(1)
+ semaphore <- struct{}{}
+ go func(partition *mq_pb.Partition, broker string) {
+ defer wg.Done()
+ defer func() { <-semaphore }()
+ glog.V(0).Infof("subscriber %s/%s/%s assigned partition %+v at %v", sub.ContentConfig.Namespace, sub.ContentConfig.Topic, sub.SubscriberConfig.ConsumerGroup, partition, broker)
+ err := sub.onEachPartition(partition, broker)
+ if err != nil {
+ glog.V(0).Infof("subscriber %s/%s/%s partition %+v at %v: %v", sub.ContentConfig.Namespace, sub.ContentConfig.Topic, sub.SubscriberConfig.ConsumerGroup, partition, broker, err)
+ }
+ }(assigned.Partition, assigned.Broker)
+ }
+
+ wg.Wait()
+}
+
+func (sub *TopicSubscriber) onEachPartition(partition *mq_pb.Partition, broker string) error {
+ // connect to the partition broker
+ return pb.WithBrokerGrpcClient(true, broker, sub.SubscriberConfig.GrpcDialOption, func(client mq_pb.SeaweedMessagingClient) error {
+ subscribeClient, err := client.SubscribeMessage(context.Background(), &mq_pb.SubscribeMessageRequest{
+ Message: &mq_pb.SubscribeMessageRequest_Init{
+ Init: &mq_pb.SubscribeMessageRequest_InitMessage{
+ ConsumerGroup: sub.SubscriberConfig.ConsumerGroup,
+ ConsumerId: sub.SubscriberConfig.ConsumerGroupInstanceId,
+ Topic: &mq_pb.Topic{
+ Namespace: sub.ContentConfig.Namespace,
+ Name: sub.ContentConfig.Topic,
+ },
+ PartitionOffset: &mq_pb.PartitionOffset{
+ Partition: partition,
+ StartTsNs: sub.alreadyProcessedTsNs,
+ StartType: mq_pb.PartitionOffsetStartType_EARLIEST_IN_MEMORY,
+ },
+ Filter: sub.ContentConfig.Filter,
+ },
+ },
+ })
+
+ if err != nil {
+ return fmt.Errorf("create subscribe client: %v", err)
+ }
+
+ glog.V(0).Infof("subscriber %s/%s/%s connected to partition %+v at %v", sub.ContentConfig.Namespace, sub.ContentConfig.Topic, sub.SubscriberConfig.ConsumerGroup, partition, broker)
+
+ if sub.OnCompletionFunc != nil {
+ defer sub.OnCompletionFunc()
+ }
+ defer func() {
+ subscribeClient.SendMsg(&mq_pb.SubscribeMessageRequest{
+ Message: &mq_pb.SubscribeMessageRequest_Ack{
+ Ack: &mq_pb.SubscribeMessageRequest_AckMessage{
+ Sequence: 0,
+ },
+ },
+ })
+ subscribeClient.CloseSend()
+ }()
+
+ for {
+ // glog.V(0).Infof("subscriber %s/%s/%s waiting for message", sub.ContentConfig.Namespace, sub.ContentConfig.Topic, sub.SubscriberConfig.ConsumerGroup)
+ resp, err := subscribeClient.Recv()
+ if err != nil {
+ return fmt.Errorf("subscribe recv: %v", err)
+ }
+ if resp.Message == nil {
+ glog.V(0).Infof("subscriber %s/%s/%s received nil message", sub.ContentConfig.Namespace, sub.ContentConfig.Topic, sub.SubscriberConfig.ConsumerGroup)
+ continue
+ }
+ switch m := resp.Message.(type) {
+ case *mq_pb.SubscribeMessageResponse_Data:
+ shouldContinue, processErr := sub.OnEachMessageFunc(m.Data.Key, m.Data.Value)
+ if processErr != nil {
+ return fmt.Errorf("process error: %v", processErr)
+ }
+ sub.alreadyProcessedTsNs = m.Data.TsNs
+ if !shouldContinue {
+ return nil
+ }
+ case *mq_pb.SubscribeMessageResponse_Ctrl:
+ // glog.V(0).Infof("subscriber %s/%s/%s received control %+v", sub.ContentConfig.Namespace, sub.ContentConfig.Topic, sub.SubscriberConfig.ConsumerGroup, m.Ctrl)
+ if m.Ctrl.IsEndOfStream || m.Ctrl.IsEndOfTopic {
+ return io.EOF
+ }
+ }
+ }
+
+ return nil
+ })
+}
diff --git a/weed/mq/client/sub_client/lookup.go b/weed/mq/client/sub_client/lookup.go
deleted file mode 100644
index b6d2a8c53..000000000
--- a/weed/mq/client/sub_client/lookup.go
+++ /dev/null
@@ -1,34 +0,0 @@
-package sub_client
-
-import (
- "context"
- "fmt"
- "github.com/seaweedfs/seaweedfs/weed/pb"
- "github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
-)
-
-func (sub *TopicSubscriber) doLookup(brokerAddress string) error {
- err := pb.WithBrokerGrpcClient(true,
- brokerAddress,
- sub.SubscriberConfig.GrpcDialOption,
- func(client mq_pb.SeaweedMessagingClient) error {
- lookupResp, err := client.LookupTopicBrokers(context.Background(),
- &mq_pb.LookupTopicBrokersRequest{
- Topic: &mq_pb.Topic{
- Namespace: sub.ContentConfig.Namespace,
- Name: sub.ContentConfig.Topic,
- },
- IsForPublish: false,
- })
- if err != nil {
- return err
- }
- sub.brokerPartitionAssignments = lookupResp.BrokerPartitionAssignments
- return nil
- })
-
- if err != nil {
- return fmt.Errorf("lookup topic %s/%s: %v", sub.ContentConfig.Namespace, sub.ContentConfig.Topic, err)
- }
- return nil
-}
diff --git a/weed/mq/client/sub_client/process.go b/weed/mq/client/sub_client/process.go
deleted file mode 100644
index b6bdb14ee..000000000
--- a/weed/mq/client/sub_client/process.go
+++ /dev/null
@@ -1,85 +0,0 @@
-package sub_client
-
-import (
- "context"
- "fmt"
- "github.com/seaweedfs/seaweedfs/weed/pb"
- "github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
- "sync"
-)
-
-func (sub *TopicSubscriber) doProcess() error {
- var wg sync.WaitGroup
- for _, brokerPartitionAssignment := range sub.brokerPartitionAssignments {
- brokerAddress := brokerPartitionAssignment.LeaderBroker
- grpcConnection, err := pb.GrpcDial(context.Background(), brokerAddress, true, sub.SubscriberConfig.GrpcDialOption)
- if err != nil {
- return fmt.Errorf("dial broker %s: %v", brokerAddress, err)
- }
- brokerClient := mq_pb.NewSeaweedMessagingClient(grpcConnection)
- subscribeClient, err := brokerClient.Subscribe(context.Background(), &mq_pb.SubscribeRequest{
- Message: &mq_pb.SubscribeRequest_Init{
- Init: &mq_pb.SubscribeRequest_InitMessage{
- ConsumerGroup: sub.SubscriberConfig.GroupId,
- ConsumerId: sub.SubscriberConfig.GroupInstanceId,
- Topic: &mq_pb.Topic{
- Namespace: sub.ContentConfig.Namespace,
- Name: sub.ContentConfig.Topic,
- },
- Partition: &mq_pb.Partition{
- RingSize: brokerPartitionAssignment.Partition.RingSize,
- RangeStart: brokerPartitionAssignment.Partition.RangeStart,
- RangeStop: brokerPartitionAssignment.Partition.RangeStop,
- },
- Filter: sub.ContentConfig.Filter,
- Offset: &mq_pb.SubscribeRequest_InitMessage_StartTimestampNs{
- StartTimestampNs: sub.alreadyProcessedTsNs,
- },
- },
- },
- })
- if err != nil {
- return fmt.Errorf("create subscribe client: %v", err)
- }
- wg.Add(1)
- go func() {
- defer wg.Done()
- if sub.OnCompletionFunc != nil {
- defer sub.OnCompletionFunc()
- }
- defer func() {
- subscribeClient.SendMsg(&mq_pb.SubscribeRequest{
- Message: &mq_pb.SubscribeRequest_Ack{
- Ack: &mq_pb.SubscribeRequest_AckMessage{
- Sequence: 0,
- },
- },
- })
- subscribeClient.CloseSend()
- }()
- for {
- resp, err := subscribeClient.Recv()
- if err != nil {
- fmt.Printf("subscribe error: %v\n", err)
- return
- }
- if resp.Message == nil {
- continue
- }
- switch m := resp.Message.(type) {
- case *mq_pb.SubscribeResponse_Data:
- if !sub.OnEachMessageFunc(m.Data.Key, m.Data.Value) {
- return
- }
- sub.alreadyProcessedTsNs = m.Data.TsNs
- case *mq_pb.SubscribeResponse_Ctrl:
- if m.Ctrl.IsEndOfStream || m.Ctrl.IsEndOfTopic {
- return
- }
- }
- }
- }()
- }
- wg.Wait()
- return nil
-}
diff --git a/weed/mq/client/sub_client/subscribe.go b/weed/mq/client/sub_client/subscribe.go
index 370f5aa3c..df62ea674 100644
--- a/weed/mq/client/sub_client/subscribe.go
+++ b/weed/mq/client/sub_client/subscribe.go
@@ -1,48 +1,11 @@
package sub_client
-import (
- "fmt"
- "github.com/seaweedfs/seaweedfs/weed/util"
- "io"
- "log"
- "time"
-)
-
// Subscribe subscribes to a topic's specified partitions.
// If a partition is moved to another broker, the subscriber will automatically reconnect to the new broker.
func (sub *TopicSubscriber) Subscribe() error {
- index := -1
- util.RetryUntil("subscribe", func() error {
- index++
- index = index % len(sub.bootstrapBrokers)
- // ask balancer for brokers of the topic
- if err := sub.doLookup(sub.bootstrapBrokers[index]); err != nil {
- return fmt.Errorf("lookup topic %s/%s: %v", sub.ContentConfig.Namespace, sub.ContentConfig.Topic, err)
- }
- if len(sub.brokerPartitionAssignments) == 0 {
- if sub.waitForMoreMessage {
- time.Sleep(1 * time.Second)
- return fmt.Errorf("no broker partition assignments")
- } else {
- return nil
- }
- }
- // treat the first broker as the topic leader
- // connect to the leader broker
+ // loop forever
+ sub.doKeepConnectedToSubCoordinator()
- // subscribe to the topic
- if err := sub.doProcess(); err != nil {
- return fmt.Errorf("subscribe topic %s/%s: %v", sub.ContentConfig.Namespace, sub.ContentConfig.Topic, err)
- }
- return nil
- }, func(err error) bool {
- if err == io.EOF {
- log.Printf("subscriber %s/%s: %v", sub.ContentConfig.Namespace, sub.ContentConfig.Topic, err)
- sub.waitForMoreMessage = false
- return false
- }
- return true
- })
return nil
}
diff --git a/weed/mq/client/sub_client/subscriber.go b/weed/mq/client/sub_client/subscriber.go
index 9b96b14cb..9adc5197e 100644
--- a/weed/mq/client/sub_client/subscriber.go
+++ b/weed/mq/client/sub_client/subscriber.go
@@ -7,10 +7,10 @@ import (
)
type SubscriberConfiguration struct {
- ClientId string
- GroupId string
- GroupInstanceId string
- GroupMinimumPeers int32
+ ClientId string
+ ConsumerGroup string
+ ConsumerGroupInstanceId string
+ GroupMinimumPeers int32
GroupMaximumPeers int32
BootstrapServers []string
GrpcDialOption grpc.DialOption
@@ -23,12 +23,17 @@ type ContentConfiguration struct {
StartTime time.Time
}
-type OnEachMessageFunc func(key, value []byte) (shouldContinue bool)
+type ProcessorConfiguration struct {
+ ConcurrentPartitionLimit int // how many partitions to process concurrently
+}
+
+type OnEachMessageFunc func(key, value []byte) (shouldContinue bool, err error)
type OnCompletionFunc func()
type TopicSubscriber struct {
SubscriberConfig *SubscriberConfiguration
ContentConfig *ContentConfiguration
+ ProcessorConfig *ProcessorConfiguration
brokerPartitionAssignments []*mq_pb.BrokerPartitionAssignment
OnEachMessageFunc OnEachMessageFunc
OnCompletionFunc OnCompletionFunc
@@ -37,10 +42,11 @@ type TopicSubscriber struct {
alreadyProcessedTsNs int64
}
-func NewTopicSubscriber(bootstrapBrokers []string, subscriber *SubscriberConfiguration, content *ContentConfiguration) *TopicSubscriber {
+func NewTopicSubscriber(bootstrapBrokers []string, subscriber *SubscriberConfiguration, content *ContentConfiguration, processor ProcessorConfiguration) *TopicSubscriber {
return &TopicSubscriber{
SubscriberConfig: subscriber,
ContentConfig: content,
+ ProcessorConfig: &processor,
bootstrapBrokers: bootstrapBrokers,
waitForMoreMessage: true,
alreadyProcessedTsNs: content.StartTime.UnixNano(),
generated by cgit v1.2.3 (git 2.25.1) at 2025-12-24 15:25:35 +0000