You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "panszobe (via GitHub)" <gi...@apache.org> on 2024/02/23 08:31:30 UTC
[I] Consumers can't keep on consuming messages while Pulsar Broker restarting even the consumer has been reconnected to the broker [pulsar-client-go]
panszobe opened a new issue, #1180:
URL: https://github.com/apache/pulsar-client-go/issues/1180
#### Expected behavior
While Pulsar Broker restarting, consumers which have reconnected to broker should keep on consuming messages by sending the flow requests to Pulsar Server. And the Pulsar Broker should keep on pushing messages to Client.
#### Actual behavior
We discovered that our consumer would not consume new messages after reconnected to the broker, the monitoring metrics graph as below:
Server Side: Bytes Out per Subscription (bytes/s) (group by partitioned-topic)
Client Side: consumed rows/s
Since 14:40, the output decrease to 0.
We did a case check which indicated that Client could not keep on sending the flow requests to Server, so Server would not push messages to Client because permits < 0.
The log of Server as below:
```
14:36:42.142 [pulsar-io-32-25] INFO org.apache.pulsar.broker.service.ServerCnx - [/xxxx:55102] Subscribing on topic persistent://test/normal/test_consume_duplicated_for_ck_sinker-partition-2 / ck_sinker_preonline_sinker_current_version_test_consume_for_ck_sinker_pulsar_v_origin
14:36:42.206 [pulsar-io-32-25] DEBUG org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers - [persistent://test/normal/test_consume_duplicated_for_ck_sinker-partition-2 / ck_sinker_preonline_sinker_current_version_test_consume_for_ck_sinker_pulsar_v_origin-Consumer{subscription=PersistentSubscription{topic=persistent://test/normal/test_consume_duplicated_for_ck_sinker-partition-2, name=ck_sinker_preonline_sinker_current_version_test_consume_for_ck_sinker_pulsar_v_origin}, consumerId=3, consumerName=vyvwu, address=/xxxx:55102}] Trigger new read after receiving flow control message with permits 1000 after adding 1000 permits
14:36:42.219 [BookKeeperClientWorker-OrderedExecutor-8-0] DEBUG org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers - [persistent://test/normal/test_consume_duplicated_for_ck_sinker-partition-2 / ck_sinker_preonline_sinker_current_version_test_consume_for_ck_sinker_pulsar_v_origin] Distributing 1 messages to 1 consumers
14:36:42.219 [BookKeeperClientWorker-OrderedExecutor-8-0] DEBUG org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers - [persistent://test/normal/test_consume_duplicated_for_ck_sinker-partition-2 / ck_sinker_preonline_sinker_current_version_test_consume_for_ck_sinker_pulsar_v_origin] Added -(2346 minus 0) permits to TOTAL_AVAILABLE_PERMITS_UPDATER in PersistentDispatcherMultipleConsumers
14:36:42.219 [pulsar-io-32-40] DEBUG org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers - [persistent://test/normal/test_consume_duplicated_for_ck_sinker-partition-2 / ck_sinker_preonline_sinker_current_version_test_consume_for_ck_sinker_pulsar_v_origin] Consumer buffer is full, pause reading
14:36:42.490 [pulsar-io-32-25] DEBUG org.apache.pulsar.broker.service.Consumer - [persistent://test/normal/test_consume_duplicated_for_ck_sinker-partition-2-PersistentSubscription{topic=persistent://test/normal/test_consume_duplicated_for_ck_sinker-partition-2, name=ck_sinker_preonline_sinker_current_version_test_consume_for_ck_sinker_pulsar_v_origin}] Added more flow control message permits 500 (old was: -1346), blocked = false
14:36:42.493 [pulsar-io-32-25] DEBUG org.apache.pulsar.broker.service.Consumer - [persistent://test/normal/test_consume_duplicated_for_ck_sinker-partition-2-PersistentSubscription{topic=persistent://test/normal/test_consume_duplicated_for_ck_sinker-partition-2, name=ck_sinker_preonline_sinker_current_version_test_consume_for_ck_sinker_pulsar_v_origin}] Added more flow control message permits 500 (old was: -846), blocked = false
14:36:42.498 [pulsar-io-32-25] DEBUG org.apache.pulsar.broker.service.Consumer - [persistent://test/normal/test_consume_duplicated_for_ck_sinker-partition-2-PersistentSubscription{topic=persistent://test/normal/test_consume_duplicated_for_ck_sinker-partition-2, name=ck_sinker_preonline_sinker_current_version_test_consume_for_ck_sinker_pulsar_v_origin}] Added more flow control message permits 500 (old was: -346), blocked = false
14:36:42.501 [pulsar-io-32-25] DEBUG org.apache.pulsar.broker.service.Consumer - [persistent://test/normal/test_consume_duplicated_for_ck_sinker-partition-2-PersistentSubscription{topic=persistent://test/normal/test_consume_duplicated_for_ck_sinker-partition-2, name=ck_sinker_preonline_sinker_current_version_test_consume_for_ck_sinker_pulsar_v_origin}] Added more flow control message permits 500 (old was: 154), blocked = false
14:36:42.524 [BookKeeperClientWorker-OrderedExecutor-8-0] DEBUG org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers - [persistent://test/normal/test_consume_duplicated_for_ck_sinker-partition-2 / ck_sinker_preonline_sinker_current_version_test_consume_for_ck_sinker_pulsar_v_origin] Distributing 42 messages to 1 consumers
14:36:42.524 [BookKeeperClientWorker-OrderedExecutor-8-0] DEBUG org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers - [persistent://test/normal/test_consume_duplicated_for_ck_sinker-partition-2 / ck_sinker_preonline_sinker_current_version_test_consume_for_ck_sinker_pulsar_v_origin] Added -(2223 minus 0) permits to TOTAL_AVAILABLE_PERMITS_UPDATER in PersistentDispatcherMultipleConsumers
14:36:42.535 [pulsar-io-32-25] DEBUG org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers - [persistent://test/normal/test_consume_duplicated_for_ck_sinker-partition-2 / ck_sinker_preonline_sinker_current_version_test_consume_for_ck_sinker_pulsar_v_origin-Consumer{subscription=PersistentSubscription{topic=persistent://test/normal/test_consume_duplicated_for_ck_sinker-partition-2, name=ck_sinker_preonline_sinker_current_version_test_consume_for_ck_sinker_pulsar_v_origin}, consumerId=3, consumerName=vyvwu, address=/xxxx:55102}] Trigger new read after receiving flow control message with permits -1569 after adding 0 permits
14:36:42.524 [pulsar-io-32-29] DEBUG org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers - [persistent://test/normal/test_consume_duplicated_for_ck_sinker-partition-2 / ck_sinker_preonline_sinker_current_version_test_consume_for_ck_sinker_pulsar_v_origin] Consumer buffer is full, pause reading
```
The log of Clinet as below(some debugging infomation added):
```
time="2024-02-23T14:36:42+08:00" level=info msg="Reconnecting to broker in 113.752291ms" consumerID=3 name=vyvwu subscription=ck_sinker_preonline_sinker_current_version_test_consume_for_ck_sinker_pulsar_v_origin topic="persistent://test/normal/test_consume_duplicated_for_ck_sinker-partition-2"
time="2024-02-23T14:36:42+08:00" level=info msg="Connected consumer" consumerID=3 name=vyvwu subscription=ck_sinker_preonline_sinker_current_version_test_consume_for_ck_sinker_pulsar_v_origin topic="persistent://test/normal/test_consume_duplicated_for_ck_sinker-partition-2"
time="2024-02-23T14:36:42+08:00" level=info msg="Reconnected consumer to broker" consumerID=3 name=vyvwu subscription=ck_sinker_preonline_sinker_current_version_test_consume_for_ck_sinker_pulsar_v_origin topic="persistent://test/normal/test_consume_duplicated_for_ck_sinker-partition-2"
time="2024-02-23T14:36:42+08:00" level=info msg="dispatcher requesting initial permits=1000, message stats: current total entries=0, received messages=0, duplicated messages=0, acked messages=0, discarded messages=0, total skipped messages=0" consumerID=3 name=vyvwu subscription=ck_sinker_preonline_sinker_current_version_test_consume_for_ck_sinker_pulsar_v_origin topic="persistent://test/normal/test_consume_duplicated_for_ck_sinker-partition-2"
time="2024-02-23T14:36:42+08:00" level=info msg="requesting more permits=500 available=500, message stats: current total entries=1, received messages=500, duplicated messages=0, acked messages=0, discarded messages=0, total skipped messages=0" consumerID=3 name=vyvwu subscription=ck_sinker_preonline_sinker_current_version_test_consume_for_ck_sinker_pulsar_v_origin topic="persistent://test/normal/test_consume_duplicated_for_ck_sinker-partition-2"
time="2024-02-23T14:36:42+08:00" level=info msg="requesting more permits=500 available=500, message stats: current total entries=1, received messages=1000, duplicated messages=0, acked messages=0, discarded messages=0, total skipped messages=0" consumerID=3 name=vyvwu subscription=ck_sinker_preonline_sinker_current_version_test_consume_for_ck_sinker_pulsar_v_origin topic="persistent://test/normal/test_consume_duplicated_for_ck_sinker-partition-2"
time="2024-02-23T14:36:42+08:00" level=info msg="requesting more permits=500 available=500, message stats: current total entries=1, received messages=1500, duplicated messages=0, acked messages=0, discarded messages=0, total skipped messages=0" consumerID=3 name=vyvwu subscription=ck_sinker_preonline_sinker_current_version_test_consume_for_ck_sinker_pulsar_v_origin topic="persistent://test/normal/test_consume_duplicated_for_ck_sinker-partition-2"
time="2024-02-23T14:36:42+08:00" level=info msg="requesting more permits=500 available=500, message stats: current total entries=1, received messages=2000, duplicated messages=0, acked messages=0, discarded messages=0, total skipped messages=0" consumerID=3 name=vyvwu subscription=ck_sinker_preonline_sinker_current_version_test_consume_for_ck_sinker_pulsar_v_origin topic="persistent://test/normal/test_consume_duplicated_for_ck_sinker-partition-2"
time="2024-02-23T14:38:19+08:00" level=info msg="message stats: current total entries=2, received messages=4569, duplicated messages=2223, acked messages=0, discarded messages=0, total skipped messages=2223" consumerID=3 name=vyvwu subscription=ck_sinker_preonline_sinker_current_version_test_consume_for_ck_sinker_pulsar_v_origin topic="persistent://test/normal/test_consume_duplicated_for_ck_sinker-partition-2"
time="2024-02-23T14:40:19+08:00" level=info msg="message stats: current total entries=2, received messages=4569, duplicated messages=2223, acked messages=0, discarded messages=0, total skipped messages=2223" consumerID=3 name=vyvwu subscription=ck_sinker_preonline_sinker_current_version_test_consume_for_ck_sinker_pulsar_v_origin topic="persistent://test/normal/test_consume_duplicated_for_ck_sinker-partition-2"
time="2024-02-23T14:42:19+08:00" level=info msg="message stats: current total entries=2, received messages=4569, duplicated messages=2223, acked messages=0, discarded messages=0, total skipped messages=2223" consumerID=3 name=vyvwu subscription=ck_sinker_preonline_sinker_current_version_test_consume_for_ck_sinker_pulsar_v_origin topic="persistent://test/normal/test_consume_duplicated_for_ck_sinker-partition-2"
```
Client sended 5 flow requests to Server and Server handled 5 times as excepted, but there are 2223 messages duplicated so that Client will drop these messages in function _MessageReceived()_ which will not increase the avalialePermits when skipping some messages.
As a result, avalialePermits of such partitionConsumer not reached to 500(maxQueueSize/2, use default maxQueueSize=1000), current avaliablePermits = (4569 - 2223) % 500 = 346.
#### Steps to reproduce
Producers: using batch by setting DisableBatching=false, MaxPendingMessages = 100000, BatchingMaxSize = 1073741824
Consumers: setting EnableBatchIndexAcknowledgment = true
When batchSize is much more greater, the effects of reproduction are more clear.
#### Probable Bugs
https://github.com/apache/pulsar-client-go/blob/3b9b1f8895d8924ec98db4612806b9871f1d135b/pulsar/consumer_partition.go#L1103-L1106
https://github.com/apache/pulsar-client-go/blob/3b9b1f8895d8924ec98db4612806b9871f1d135b/pulsar/consumer_partition.go#L1121-L1124
https://github.com/apache/pulsar-client-go/blob/3b9b1f8895d8924ec98db4612806b9871f1d135b/pulsar/consumer_partition.go#L1146-L1148
These will not increase the availablePermits.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Consumers can't keep on consuming messages while Pulsar Broker restarting even the consumer has been reconnected to the broker [pulsar-client-go]
Posted by "RobertIndie (via GitHub)" <gi...@apache.org>.
RobertIndie closed issue #1180: Consumers can't keep on consuming messages while Pulsar Broker restarting even the consumer has been reconnected to the broker
URL: https://github.com/apache/pulsar-client-go/issues/1180
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org