diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java index ddc955f84d82b..959bbe6a80fdf 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java @@ -41,6 +41,8 @@ import io.netty.buffer.ByteBuf; import java.io.InputStream; import java.lang.reflect.Field; +import java.net.ServerSocket; +import java.net.URI; import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; import java.security.GeneralSecurityException; @@ -48,6 +50,7 @@ import java.util.HashSet; import java.util.List; import java.util.NavigableMap; +import java.util.Optional; import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentSkipListMap; @@ -1128,4 +1131,43 @@ public void testUnsubscribeForce(SubscriptionType type) throws Exception { consumer1.unsubscribe(true); assertFalse(consumer1.isConnected()); } + + /** + * Regression test for https://github.com/apache/pulsar/issues/25997: a broker-assigned redirect + * ({@code CloseProducer}/{@code CloseConsumer} carrying an {@code assignedBrokerServiceUrl}) + * pointing at a wrong or unreachable broker must be honored for the immediate reconnect attempt + * only. If that attempt fails, subsequent retries must fall back to topic lookup instead of + * staying pinned to the stale address. + */ + @Test + public void testStaleBrokerRedirectFallsBackToLookup() throws Exception { + String topic = "persistent://my-property/my-ns/staleRedirectFallback"; + @Cleanup + Producer producerInstance = pulsarClient.newProducer().topic(topic).create(); + @Cleanup + Consumer consumerInstance = pulsarClient.newConsumer().topic(topic) + .subscriptionName("my-subscriber-name").subscribe(); + ProducerImpl producer = (ProducerImpl) producerInstance; + ConsumerImpl consumer = (ConsumerImpl) consumerInstance; + + // An address with no listener behind it: dialing the redirect fails immediately. + URI unreachable; + try (ServerSocket socket = new ServerSocket(0)) { + unreachable = URI.create("pulsar://127.0.0.1:" + socket.getLocalPort()); + } + + // Deliver the disconnect+redirect exactly as ClientCnx#handleCloseProducer and + // #handleCloseConsumer do for a broker unload notification. + producer.connectionClosed(producer.getClientCnx(), Optional.of(0L), Optional.of(unreachable)); + consumer.connectionClosed(consumer.getClientCnx(), Optional.of(0L), Optional.of(unreachable)); + + // The redirect dial fails; the clients must recover by looking up the topic again. + Awaitility.await().atMost(15, TimeUnit.SECONDS).untilAsserted(() -> { + assertEquals(producer.getState(), State.Ready); + assertEquals(consumer.getState(), State.Ready); + }); + + producer.send("msg".getBytes(UTF_8)); + assertNotNull(consumer.receive(10, TimeUnit.SECONDS)); + } } diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionHandler.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionHandler.java index e398b0583fa8b..5538feb85374d 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionHandler.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionHandler.java @@ -227,9 +227,15 @@ public void connectionClosed(ClientCnx cnx) { public void connectionClosed(ClientCnx cnx, Optional initialConnectionDelayMs, Optional hostUrl) { lastConnectionClosedTimestamp = System.currentTimeMillis(); duringConnect.set(false); - // Remember an explicit reconnect target so a later first-attempt failure (reconnectLater) - // re-dials the same broker rather than falling back to the service URL. - hostUrl.ifPresent(uri -> this.explicitHostURI = uri); + // Only handlers already pinned to an explicit host (v5 TC metadata-store discovery) update + // the pin here, so a later first-attempt failure (reconnectLater) re-dials the current + // leader rather than falling back to the service URL. For lookup-based handlers + // (producers/consumers), the broker-assigned redirect in hostUrl may be wrong or stale: + // it is honored for the immediate reconnect attempt below only, and retries after a + // failure go through a fresh topic lookup instead of staying pinned to it. + if (explicitHostURI != null) { + hostUrl.ifPresent(uri -> this.explicitHostURI = uri); + } state.client.getCnxPool().releaseConnection(cnx); if (CLIENT_CNX_UPDATER.compareAndSet(this, cnx, null)) { if (!state.changeToConnecting()) {