https://pulsar.apache.org/ logo
Join Slack
Powered by
Question regarding best practice when dealing with...
# general
p
Question regarding best practice when dealing with subscribers in multiple regions. I have 2 clusters, West and East. I have 1 pub in West. 2 subs in West and East. If I publish 10 messages, if both subs are online they will receive these 10 messages. However, if I drop 1 sub. 1 sub will receive the 10 messages and acknowledge them. 2nd sub comes online and there are no messages to process. Do I need to manually keep track of the message IDs myself? So that if a subscriber goes down and then comes back up later on. They’ll process messages from when they last acknowledged? Finally, if I have a cluster of subscribers. Do I try to track a global last message Id across the geo-location. Or do I track against an individual subscriber and then handle duplications myself? Thanks!
e
🙌 1
p
Thank you!
So I’m not running a global zookeeper and I went through the video. At no point is a subscriber not running and a publisher sending messages.
e
p
Looking at the links provided. You are giving me instructions on how to set this all up?
e
Yes! So that you can learn about it - is it your first time?
README.md
p
Yes, it’s my first time. But I already have working clusters with geo-replication. 😄
e
I am showing you examples; this is c#
p
Thanks but I think I worked it out. All I needed was a reader proceeding a subscriber
e
subs
must be the same for geo-repl
replication.webp
d
@Paul M, do you have a message retention policy configured on the topic you are consuming from? In short, if all of the known subscribers on a topic have acked the messages, they are deleted unless you specify a retention policy to keep them.
p
@David K Not yet. I figure my retention policy would be an hour at most. As given there is a cluster of about 3-5 subscribers per region running. But I’m coming at this from a DR scenario. I have found a way to get a worker to be a reader and then a subscriber. But I’m struggling to get the last acknowledged messageId for the cluster.
All I have to go on is:
Copy code
String topic = "<persistent://my-tenant/my-namespace/my-topic>";
admin.topics().getLastMessage(topic);
And the golang source isn’t much help.
d
I would recommend setting your retention policy now so you don’t run into the scenario where some consumers get messages and others don’t. If the messages aren’t retained, you won’t be able to read them either.
p
I’m running a poc right now. And I’m able to rewind at various points fine. The messages have been acknowledged but not deleted as of yet.
d
Probably because they are all inside the same physical segment….. 😃 Once you increase the data volume, new segments will be created, and the older ones will be deleted.
p
Understood!
After a bit of digging, got the last acknowledged messageId.
Copy code
package main

import (
	"fmt"

	"<http://github.com/apache/pulsar-client-go/pulsaradmin|github.com/apache/pulsar-client-go/pulsaradmin>"
	"<http://github.com/apache/pulsar-client-go/pulsaradmin/pkg/utils|github.com/apache/pulsar-client-go/pulsaradmin/pkg/utils>"
)

func main() {
	cfg := &pulsaradmin.Config{
		WebServiceURL: "<http://10.211.55.10:8080>",
	}
	admin, err := pulsaradmin.NewClient(cfg)
	if err != nil {
		panic(err)
	}

	topicName, err := utils.GetTopicName("my-tenant/my-namespace/my-topic")

	if err != nil {
		panic(err)
	}

	messageId, err := admin.Topics().GetLastMessageID(*topicName)

	if err != nil {
		panic(err)
	}

	fmt.Println(messageId)
}
👍 1
10 Views
Open in Slack
PreviousNext
Powered by