@ -85,7 +85,7 @@ func (sub *TopicSubscriber) onEachAssignment(assignment *mq_pb.SubscriberToSubCo
go func ( partition * mq_pb . Partition , broker string ) {
go func ( partition * mq_pb . Partition , broker string ) {
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 ( 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 )
err := sub . onEachPartition ( partition , broker )
if err != nil {
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 )
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 )
@ -125,7 +125,7 @@ func (sub *TopicSubscriber) onEachPartition(partition *mq_pb.Partition, broker s
return fmt . Errorf ( "create subscribe client: %v" , err )
return fmt . Errorf ( "create subscribe client: %v" , err )
}
}
glog . V ( 1 ) . Infof ( "subscriber %s/%s/%s connected to partition %+v at %v" , sub . ContentConfig . Namespace , sub . ContentConfig . Topic , sub . SubscriberConfig . ConsumerGroup , partition , broker )
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 {
if sub . OnCompletionFunc != nil {
defer sub . OnCompletionFunc ( )
defer sub . OnCompletionFunc ( )