gitlab-com-seaweedfs/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2024-01-19 02:48:24 +00:00
Code Issues Projects Releases Wiki Activity

log errors

Browse source
This commit is contained in:
chrislu 2024-01-01 19:09:17 -08:00
parent bf9ae74384
commit 8b4cd50aac
1 changed files with 6 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
weed/mq/client/sub_client/connect_to_sub_coordinator.go
View file

@ -86,16 +86,19 @@ func (sub *TopicSubscriber) onEachAssignment(assignment *mq_pb.SubscriberToSubCo
defer wg.Done() defer wg.Done()
defer func() { <-semaphore }() defer func() { <-semaphore }()
glog.V(1).Infof("subscriber %s/%s/%s assigned partition %+v at %v", sub.ContentConfig.Namespace, sub.ContentConfig.Topic, sub.SubscriberConfig.ConsumerGroup, partition, broker) glog.V(1).Infof("subscriber %s/%s/%s assigned partition %+v at %v", sub.ContentConfig.Namespace, sub.ContentConfig.Topic, sub.SubscriberConfig.ConsumerGroup, partition, broker)
sub.onEachPartition(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) }(assigned.Partition, assigned.Broker)
} }
wg.Wait() wg.Wait()
} }
func (sub *TopicSubscriber) onEachPartition(partition *mq_pb.Partition, broker string) { func (sub *TopicSubscriber) onEachPartition(partition *mq_pb.Partition, broker string) error {
// connect to the partition broker // connect to the partition broker
pb.WithBrokerGrpcClient(true, broker, sub.SubscriberConfig.GrpcDialOption, func(client mq_pb.SeaweedMessagingClient) error { return pb.WithBrokerGrpcClient(true, broker, sub.SubscriberConfig.GrpcDialOption, func(client mq_pb.SeaweedMessagingClient) error {
subscribeClient, err := client.Subscribe(context.Background(), &mq_pb.SubscribeRequest{ subscribeClient, err := client.Subscribe(context.Background(), &mq_pb.SubscribeRequest{
Message: &mq_pb.SubscribeRequest_Init{ Message: &mq_pb.SubscribeRequest_Init{
Init: &mq_pb.SubscribeRequest_InitMessage{ Init: &mq_pb.SubscribeRequest_InitMessage{