Browse Source

stop partitionOffsetChan if closed

pull/5890/head
chrislu 7 months ago
parent
commit
23a72db1df
  1. 5
      weed/mq/client/sub_client/on_each_partition.go

5
weed/mq/client/sub_client/on_each_partition.go

@ -67,6 +67,11 @@ func (sub *TopicSubscriber) onEachPartition(assigned *mq_pb.BrokerPartitionAssig
subscribeClient.CloseSend() subscribeClient.CloseSend()
return return
case ack := <-partitionOffsetChan: case ack := <-partitionOffsetChan:
case ack, ok := <-partitionOffsetChan:
if !ok {
subscribeClient.CloseSend()
return
}
subscribeClient.SendMsg(&mq_pb.SubscribeMessageRequest{ subscribeClient.SendMsg(&mq_pb.SubscribeMessageRequest{
Message: &mq_pb.SubscribeMessageRequest_Ack{ Message: &mq_pb.SubscribeMessageRequest_Ack{
Ack: &mq_pb.SubscribeMessageRequest_AckMessage{ Ack: &mq_pb.SubscribeMessageRequest_AckMessage{

Loading…
Cancel
Save