Can broker publish messages to topics in other brokers ?

[Richard-Yi]

I want to implement a custom feature on pulsar.
the feature can be simply described as below:

1. send a message to topic_1
2. generate a new message after sending complete and send the new message to topic_2

The feature works fine in standalone mode. However, after testing, I found that broker can't publish messages to topic_2 in a cluster mode, because topic_1 and topic_2 are not in the same broker.

My question is, is pulsar currently able to implement the above requirements and is there a wrapped API for it? Or it's a must that both topics must be on the same broker.

[RobertIndie]

Pulsar already supports publishing messages to topics in other brokers.

I found that broker can't publish messages to topic_2 in a cluster mode, because topic_1 and topic_2 are not in the same broker.

Are there any exceptions?
Before publishing a message, pulsar does a lookup for the topic to determine which broker the topic is on and then publishes that message to that broker.

[hezhangjian]

Have you looked at pulsar function? It might can help you reach the goal

[Richard-Yi]

@shoothzj pulsar function works in theory. But it need specify the input topics when deploy. In my situation, the feature need to work on a namespace level. I can't not specify topics value which will be added in the future.

[Richard-Yi]

@RobertIndie Can you show some example code?
My code is as follows

public CompletableFuture<Topic> getTopic(String topicName, boolean createIfMissing) {
        CompletableFuture<Topic> topicCompletableFuture = new CompletableFuture<>();
        if (null == pulsarService) {
            log.error("PulsarService is not set.");
            topicCompletableFuture.completeExceptionally(new PulsarServerException("PulsarService is not set."));
            return topicCompletableFuture;
        }
        // setup ownership of service unit to this broker
        LookupOptions lookupOptions = LookupOptions.builder().authoritative(true).build();
        pulsarService.getNamespaceService().getBrokerServiceUrlAsync(TopicName.get(topicName), lookupOptions).
                whenComplete((addr, th) -> {
                    log.info("Find getBrokerServiceUrl {}, return Topic: {}", addr, topicName);
                    if (th != null || addr == null || addr.get() == null) {
                        log.warn("Failed getBrokerServiceUrl {}, return null Topic. throwable: ", topicName, th);
                        topicCompletableFuture.complete(null);
                        return;
                    }
                    if (log.isDebugEnabled()) {
                        log.debug("getBrokerServiceUrl for {} in ExchangeTopicManager. brokerAddress: {}",
                                topicName, addr.get().getLookupData().getBrokerUrl());
                    }
                    pulsarService.getBrokerService().getTopic(topicName, createIfMissing)
                            .whenComplete((topicOptional, throwable) -> {
                                if (throwable != null) {
                                    log.error("Failed to getTopic {}. exception: {}", topicName, throwable);
                                    topicCompletableFuture.complete(null);
                                    return;
                                }
                                try {
                                    if (topicOptional.isPresent()) {
                                        Topic topic = topicOptional.get();
                                        AbstractTopic abstractTopic = (AbstractTopic) topic;
                                        abstractTopic.setDeleteWhileInactive(false);
                                        topicCompletableFuture.complete(topic);
                                    } else {
                                        log.error("Get empty topic for name {}", topicName);
                                        topicCompletableFuture.complete(null);
                                    }
                                } catch (Exception e) {
                                    log.error("Failed to get client in registerInPersistentTopic {}. "
                                            + "exception:", topicName, e);
                                    topicCompletableFuture.complete(null);
                                }
                            });
                });
        return topicCompletableFuture;
    }

after getting the topic,

 getTopic(...).whenComplete((topic, throwable) -> {
                                        if (throwable != null) {
                                            log.error("get message topic error.", throwable);
                                            return;
                                        }
                                        TypedMessageBuilderImpl<byte[]>
                                                builder = new TypedMessageBuilderImpl<>(null, Schema.BYTES);
                                        builder.value(new byte[0])
                                                .property(MESSAGE_ID,
                                                        messageIdAsString == null ? "" : messageIdAsString)
                                                .property(LEDGER_ID, String.valueOf(position.getLedgerId()))
                                                .property(EXCHANGE_TOPIC, exchangeTopicName)
                                                .property(ENTRY_ID, String.valueOf(position.getEntryId()));
                                        Message<byte[]> builderMessage = builder.getMessage();
                                        ByteBuf byteBuf = MessageConvertUtils.messageToByteBuf(builderMessage);
                                        try {
                                            topic.publishMessage(byteBuf, (e, ledgerId1, entryId1) -> {
                                                if (e != null) {
                                                    log.error("message index completed error，ledgerId:{} entryId:{}",
                                                            ledgerId1, entryId1, e);
                                                }
                                            });
                                        } finally {
                                            ReferenceCountUtil.release(byteBuf);
                                        }
                                    });

[Richard-Yi]

Are there any exceptions?

the error msg I find @RobertIndie

17:15:58.748 [ForkJoinPool.commonPool-worker-2] WARN  org.apache.pulsar.broker.namespace
.NamespaceService - Unable to find OwnedBundle for topic - [persistent://public/default/
message-index]
17:15:58.748 [ForkJoinPool.commonPool-worker-2] WARN  org.apache.pulsar.broker.service.B
rokerService - Namespace is being unloaded, cannot add topic message-index
17:15:58.748 [ForkJoinPool.commonPool-worker-2] ERROR io.streamnative.pulsar.handlers.am
qp.AmqpTopicManager - Failed to getTopic message-index. exception: org.apache.pulsar.bro
ker.service.BrokerServiceException$ServiceUnitNotReadyException: Namespace is being unlo
aded, cannot add topic message-index

[Jason918]

• send a message to topic_1
• generate a new message after sending complete and send the new message to topic_2

Why do you do this in broker, not client?

P.S. You need a producer client instance to produce message to topic_2.

[hezhangjian]

@shoothzj pulsar function works in theory. But it need specify the input topics when deploy. In my situation, the feature need to work on a namespace level. I can't not specify topics value which will be added in the future.

how about pattern subscribe?

[Richard-Yi]

P.S. You need a producer client instance to produce message to topic_2.

@Jason918 Thx for reminding. I find that PulsarService keeps a PulsarClient instance. It might help me out. I'll try it tomorrow.