This message was deleted.

what pulsar version are you using? Please also check the

proxyRoles

option on the broker config side. A suggested solution is to set

proxyRolesEnabled

to true, then try update cluster. cc @Yu Wei Sung

superUsers:
      proxyRolesEnabled: true

Thanks for the reply @Muni Chada We are using version 3.1.1 Also just curious why and how would enabling the proxyRoles config fix the connection already closed issue 🤔 We are able to produce data on the cluster without setting the property. It is just that when a load balance is triggered we run into this issue

Interesting. Could you share broker and proxy logs.

Is this warning or error?

Hi @Yu Wei Sung, this is error from producer logs.

My guess would be that there is an issue with that client recognizing and/or reacting to the broker’s reassignment of the topics.

client will need to reconnect to locate the new broker which handles this topic.

@Ruturaj has this been resolved?

Hi @Ken, We were using this reactive client for our load testing. We switched to plain java client wherein we did not face any issue. We're unblocked for now. We will be trying this reactive client again in future. Will definitely get in touch in case of any issue.

@Sandesh Shingare Can you tell me which version of the reactive client you were using when you ran into the issue?

@David K Apologies for late reply. We were using v0.5.0 during this run. Also if you can help, we have a doubt around this reactive pulsar client. There's no close method available for ReactiveMessageConsumer and ReactiveMessageSender classes, so we are not able to close these at runtime. Close method is available for Producer and Consumer objects in plain Java client. Is there a way to achieve this in reactive client?

Also if you can help, we have a doubt around this reactive pulsar client. There's no close method available for ReactiveMessageConsumer and ReactiveMessageSender classes, so we are not able to close these at runtime. Close method is available for Producer and Consumer objects in plain Java client. Is there a way to achieve this in reactive client?

@Sandesh Shingare The lifecycle is handled in a different way in the reactive client. It uses the reactive streams subscription. When the subscription is closed, the resources are released.

Got it, let me go through the shared references. Thank you for prompt help. 🙌

we are using the https://github.com/apache/pulsar-client-reactive client to produce messages.

it is necessary to configure the

ReactiveMessageSenderCache

. Please check the above IngestController example.

Hello @Lari Hotari, Your references helped a lot in making programming decisions. Wanted to understand, for a high throughput application, which method stands more performant for producer? sendOne() or sendMany()? Take an example of high throughput low latency rest application which produces single/multiple messages to pulsar topics on every request.

@Sandesh Shingare The decision to use sendOne or sendMany depends on the application use case. If your source for the messages to send is a Flux, you would naturally use sendMany. I don't think that there is a significant difference in performance (when the cache is properly configured ). There might also be a need to use sendOne (within flatMap) because of consistency reasons, even when the source is a Flux. You cannot assume a message to be sent before a message Id is returned. In some cases, this matters from consistency point of view.

Got it. Thank you. 👍

@Sandesh Shingare This advice isn't specific to Vert.x applications or even using the Pulsar Reactive client. It is necessary to design the messaging interactions in a way where it is designed for at-least-once semantics. This means designing for idempotency and also for ensuring that data consistency is retained when it is necessary. Message ordering design should also be considered (message ordering dependencies should be avoided if possible). I have put some thoughts in a different context in this section of another blog post: https://codingthestreams.com/pulsar/2023/03/02/scalable-distributed-system-principles.html#thoughts-about-data-consist[…]at-impacts-the-architecture . It's useful to use sequence diagrams to model the consistency boundaries and designing idempotency in a messaging application. I have had plans to prepare a conference talk on this subject and perhaps I'll do that this year. Data consistency with the ReactiveMessageSender means that your application logic should be prepared in a way that if the message sending never completes, the consistency of the application won't get corrupted. This is the ReactiveMessageSender interface:

public interface ReactiveMessageSender<T> {

	/**
	 * Send one message.
	 * @param messageSpec the spec of the message to send
	 * @return a publisher that will emit one message id and complete
	 */
	Mono<MessageId> sendOne(MessageSpec<T> messageSpec);

	/**
	 * Send multiple messages and get the sending results in the same order as the
	 * messages sent. The results are {@link MessageSendResult} objects composed of the
	 * message ID of the message sent and of the original message spec that was sent. A
	 * {@code correlationMetadata} can be attached to a {@link MessageSpec} and retrieved
	 * with {@link MessageSendResult#getCorrelationMetadata()} to correlate the messages
	 * sent with the results.
	 * @param messageSpecs the specs of the messages to send
	 * @return a publisher that will emit a {@link MessageSendResult} per message
	 * successfully sent in the order that they have been sent
	 */
	Flux<MessageSendResult<T>> sendMany(Publisher<MessageSpec<T>> messageSpecs);

}

For sendOne, the Mono should complete successfully. For sendMany, by default the Flux will ignore send errors and provide individual send errors in the MessageSendResult. If ordering is important, this should be configured in a different way by setting

.stopOnError(true)

on the ReactiveMessageSenderBuilder. In that case, the Flux will terminate on the first error.

/**
	 * Sets whether an error when sending an individual message of the
	 * {@link ReactiveMessageSender#sendMany(Publisher)} operation should stop the full
	 * stream or not. (default: false). If true, the original error is wrapped in a
	 * {@link ReactiveMessageSendingException} and propagated downstream. If false, a
	 * {@link MessageSendResult} is returned with the exception put in the
	 * {@code Exception} field and a {@code null} {@link MessageId} field.
	 * @param stopOnError true if the send stream should stop on error, false otherwise.
	 * @return the sender builder instance
	 */
	ReactiveMessageSenderBuilder<T> stopOnError(boolean stopOnError);

I hope this helps making progress. With at-least-once / idempotent messaging application design it is necessary that the application is fine with duplicate messages. Perhaps someone could recommend resources where this is explained in a good way? /cc @David K

@Lari Hotari We are observing this weird behaviour in our application. We are producing messages in increasing order but they are being written in decreasing order in pulsar topic. So the consumer is also consuming them in reverse order. E.g. Produce sequence: producer.send(element1) producer.send(element2) producer.send(element3) Received producer ack message id sequence: element3 -> 3720:0 element2 -> 3720:1 element1 -> 3720:2 Consumed elements sequence via SHARED consumer: element3 -> 3720:0 element2 -> 3720:1 element1 -> 3720:2 We have enabled batching at producer side. We are producing 1000 elements using a simple for loop and as per log traces all the 1000 elements are being added to same batch. What could be the reason for reverse order while sending messages inside batch to pulsar broker?

@Sandesh Shingare Are you using the Pulsar Reactive client and ReactiveMessageSender?

Yes, added that just for testing. I thought each subscription to send call inside for loop will be ordered. But it is exactly reverse. Let me verify the behaviour with flux as well.