liangyepianzhou commented on code in PR #21144:
URL: https://github.com/apache/pulsar/pull/21144#discussion_r1321041909
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConnectionPoolTest.java:
##########
@@ -80,6 +84,33 @@ public void testSingleIpAddress() throws Exception {
eventLoop.shutdownGracefully();
}
+ @Test
+ public void testSelectConnectionForSameProducer() throws Exception {
+ final String topicName =
BrokerTestUtil.newUniqueName("persistent://sample/standalone/ns/tp_");
+ admin.topics().createNonPartitionedTopic(topicName);
+ final CommandCloseProducer commandCloseProducer = new
CommandCloseProducer();
+ // 10 connection per broker.
+ final PulsarClient clientWith10ConPerBroker =
PulsarClient.builder().connectionsPerBroker(10)
+ .serviceUrl(lookupUrl.toString()).build();
+ ProducerImpl producer = (ProducerImpl)
clientWith10ConPerBroker.newProducer().topic(topicName).create();
+ commandCloseProducer.setProducerId(producer.producerId);
+ // An error will be reported when the Producer reconnects using a
different connection.
+ // If no error is reported, the same connection was used when
reconnecting.
+ for (int i = 0; i < 20; i++) {
+ // Trigger reconnect
+ producer.getClientCnx().handleCloseProducer(commandCloseProducer);
Review Comment:
Could you please help to clarify why multiple call `handleCloseProducer ` is
valid?
```
protected void handleCloseProducer(CommandCloseProducer closeProducer) {
log.info("[{}] Broker notification of Closed producer: {}",
remoteAddress, closeProducer.getProducerId());
final long producerId = closeProducer.getProducerId();
ProducerImpl<?> producer = producers.remove(producerId);
if (producer != null) {
producer.connectionClosed(this);
} else {
log.warn("Producer with id {} not found while closing producer
", producerId);
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
