https://pulsar.apache.org/ logo
Join Slack
Powered by
This message was deleted.
# general
s
This message was deleted.
m
Is there any idea for this problem?
a
Are you processing the same messages again or is it always stopping after 2 minutes with new messages?
m
It is always stopping after a few minutes (mostly in 2-3 minutes)
a
First place I would recommend checking is the consumer part of the topic stats and the topic internal stats to see if there is anything unusual for topic 1 when it has stopped consuming
✅ 1
d
Each of the consumers is reading from a different topic. Is it possible for the source topic of the second consumer to be out of data, e.g. the consumer has read all of the available data in the topic after 2 minutes?
m
@David K No, it is not possible. New data continusly come to topic.
I changed my code. Now I subscribe three topics with just one subscription:
Copy code
topicsToSub := []string{
            "topic1",
            "topic2",
            "topic3",
        }
consumer, err := pulsarClient.Subscribe(
                    pulsar.ConsumerOptions{
                        Topics:           topicsToSub,
                        SubscriptionName: "websocketServerDev2",
                        Type:             pulsar.Exclusive,
                    },
                )
Topic 1 and Topic 3 is non persistent, topic 2 is a persistent topic. After two minutes, Topic 1 consumer stops consuming.
Avaliable permits is zero when it was stopped:
Copy code
"consumers" : [ {
        "msgRateOut" : 1.766671142763563,
        "msgThroughputOut" : 487424.15162074467,
        "bytesOutCounter" : 274884284,
        "msgOutCounter" : 1000,
        "msgRateRedeliver" : 0.0,
        "messageAckRate" : 1.4666703404314243,
        "chunkedMessageRate" : 0.0,
        "consumerName" : "bfhcb",
        "availablePermits" : 0,
        "unackedMessages" : 0,
        "avgMessagesPerEntry" : 1,
        "blockedConsumerOnUnackedMsgs" : false,
        "lastAckedTimestamp" : 1701771776625,
        "lastConsumedTimestamp" : 1701771776549,
        "lastConsumedFlowTimestamp" : 1701771553526,
        "metadata" : { },
        "address" : "/172.30.240.192:40024",
        "connectedSince" : "2023-12-05T10:19:13.526500503Z",
        "clientVersion" : "Pulsar Go v0.9.1-0.20231010024926-0bf2f9b86f4e",
        "lastAckedTime" : "2023-12-05T10:22:56.625Z",
        "lastConsumedTime" : "2023-12-05T10:22:56.549Z"
      } ],
a
In addition to 
availablePermits
being zero, the 
msgOutCounter
is exactly 1000 (matching the default value of the 
receiverQueueSize
). This suggests that there is some issue with the flow command being sent. Usually, you would see that there are unack’d messages but here it says zero, very strange. How do you ack messages in your application, do you do it individually?
m
Yes, I acknowledge them by one by. This is my code structure:
Copy code
for {
	msgConsumed, err = consumer.Receive(context.Background())
	if err != nil {
		log.Fatalln("Could not read message: ", err)
	}
	msgConsumedStr := string(msgConsumed.Payload())
	// some conditional statements over incoming messages
	// send processed data to channel
	// ...
	consumer.Ack(msgConsumed)
}
a
For the error branch you could send a negative acknowledgement but beside that it looks fine to me. I do remember debugging a similar issue with @Asaf Mesika quite some time ago, maybe he remembers
🙌 1
Open in Slack
PreviousNext
Powered by