How to dynamically produce and consume topic data

[beyondyinjl2]

After creating a Producer, I want to send data to multiple different topics, or create a Consumer, I want to consume data from multiple topics

[beyondyinjl2]

For example, first subscribe to topic:persistent://public/default/my-topic:
Consumer consumer = client.newConsumer()
.topic("persistent://public/default/my-topic")
.subscriptionName("my-subscription")
.subscribe();
Then I want to dynamically subscribe to other topics: persistent://public/default/aaa or persistent://public/mynamesapce/bbb without restarting the service

[beyondyinjl2]

When producing data, I want to dynamically send it to different topics. The topics may be in different namespaces and different tenants.

[beyondyinjl2]

Like mqtt, a producer can send data on any topic, and a consumer can subscribe to topics dynamically

[liangyuanpeng]

You can use client.newConsumer().topicsPattern() for multiple topics.

[lhotari]

When producing data, I want to dynamically send it to different topics. The topics may be in different namespaces and different tenants.

In Pulsar client, you must create a separate producer for sending to a different topic. There are many ways to achieve what you are describing. The simplest approach is to create a producer, send the message and then close the producer. However, for most cases, this wouldn't be efficient, and producer caching is needed.
This challenge is handled in the Reactive Streams adapter for Pulsar Java Client, for example. However, if you aren't using Reactive Streams it would be an overkill to use the library only for producer caching.

Here is a code example in the Pulsar code base of producer caching:

pulsar/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java

Lines 148 to 170 in 18f2f4a

	protected Producer<T> getProducer(String destinationTopic, Schema schema) {
	return getProducer(destinationTopic, null, destinationTopic, schema);
	}
	protected Producer<T> getProducer(String producerId, String producerName, String topicName, Schema schema) {
	return publishProducers.computeIfAbsent(producerId, s -> {
	try {
	log.info("Initializing producer {} on topic {} with schema {}",
	producerName, topicName, schema);
	Producer<T> producer = createProducer(
	client,
	topicName,
	producerName,
	schema != null ? schema : this.schema);
	log.info("Initialized producer {} on topic {} with schema {}: {} -> {}",
	producerName, topicName, schema, producerId, producer);
	return producer;
	} catch (PulsarClientException e) {
	log.error("Failed to create Producer while doing user publish", e);
	throw new RuntimeException(e);
	}
	});
	}

The problem with that solution is that the producers accumulate and don't get closed automatically. Closing idle producers can be achieved by using Caffeine cache as the "container". Caffeine's removalListener feature can be used to close the producer before disposing it.

I hope this helps implementing your use case.

[github-actions]

The issue had no activity for 30 days, mark with Stale label.