You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/06/01 20:09:35 UTC
(pulsar) branch branch-4.0 updated (b34fff7a19b -> 182d8858bd9)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from b34fff7a19b [fix][broker] Fix non-batched null-value messages not removed during topic compaction (#25817)
new bb0b7d5a776 [improve][client] Best-effort retry for individual/batch-index acks on send failure when ackReceiptEnabled=false (#25525)
new 644b55bee98 [fix][broker] Fix precision loss in DataSketchesSummaryLogger by replacing LongAdder with DoubleAdder for sum accumulation (#25594)
new 0693715ca09 [fix][broker] Fix stuck chunks in SharedConsumerAssignor permit tracking (#25620)
new 2b9ef9239dd [fix][client] Fix stale Healthy state in SameAuthParamsLookupAutoClusterFailover causing flaky test (#25388)
new 182d8858bd9 [fix][client] Reset higher-index states on recovery in SameAuthParamsLookupAutoClusterFailover (#25826)
The 5 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../broker/service/SharedConsumerAssignor.java | 10 +-
.../metrics/DataSketchesSummaryLogger.java | 7 +-
.../org/apache/pulsar/client/impl/ClientCnx.java | 26 +-
.../PersistentAcknowledgmentsGroupingTracker.java | 75 ++++-
.../SameAuthParamsLookupAutoClusterFailover.java | 16 ++
.../impl/AcknowledgementsGroupingTrackerTest.java | 94 +++++-
.../impl/ClientCnxRequestTimeoutQueueTest.java | 1 +
.../pulsar/client/impl/ClientTestFixtures.java | 5 +
...ameAuthParamsLookupAutoClusterFailoverTest.java | 317 +++++++++++++++++++++
9 files changed, 516 insertions(+), 35 deletions(-)
create mode 100644 pulsar-client/src/test/java/org/apache/pulsar/client/impl/SameAuthParamsLookupAutoClusterFailoverTest.java
(pulsar) 04/05: [fix][client] Fix stale Healthy state in SameAuthParamsLookupAutoClusterFailover causing flaky test (#25388)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 2b9ef9239ddb83b5baa175ed2ba0b4687b1acaac
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Mon Mar 23 17:11:17 2026 +0200
[fix][client] Fix stale Healthy state in SameAuthParamsLookupAutoClusterFailover causing flaky test (#25388)
Co-authored-by: Claude Opus 4.6 (1M context) <no...@anthropic.com>
(cherry picked from commit 3bc834f2fa8aae1ffe8f2fb3f795acd3d0de755d)
---
.../SameAuthParamsLookupAutoClusterFailover.java | 5 +
...ameAuthParamsLookupAutoClusterFailoverTest.java | 254 +++++++++++++++++++++
2 files changed, 259 insertions(+)
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/SameAuthParamsLookupAutoClusterFailover.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/SameAuthParamsLookupAutoClusterFailover.java
index 743f2e15164..5ae7b96906e 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/SameAuthParamsLookupAutoClusterFailover.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/SameAuthParamsLookupAutoClusterFailover.java
@@ -143,6 +143,11 @@ public class SameAuthParamsLookupAutoClusterFailover implements ServiceUrlProvid
for (int i = currentPulsarServiceIndex + 1; i < pulsarServiceUrlArray.length; i++) {
if (probeAvailable(i)) {
return i;
+ } else {
+ // Mark the service as Failed to prevent a spurious recovery to it
+ // after we failover to a higher-indexed service.
+ pulsarServiceStateArray[i] = PulsarServiceState.Failed;
+ checkCounterArray[i].setValue(0);
}
}
return -1;
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/SameAuthParamsLookupAutoClusterFailoverTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/SameAuthParamsLookupAutoClusterFailoverTest.java
new file mode 100644
index 00000000000..31e2bf6aaed
--- /dev/null
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/SameAuthParamsLookupAutoClusterFailoverTest.java
@@ -0,0 +1,254 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.impl;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import io.netty.channel.EventLoopGroup;
+import java.lang.reflect.Method;
+import java.net.InetSocketAddress;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import org.apache.commons.lang3.mutable.MutableInt;
+import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.pulsar.client.impl.SameAuthParamsLookupAutoClusterFailover.PulsarServiceState;
+import org.apache.pulsar.client.util.ExecutorProvider;
+import org.apache.pulsar.common.util.FutureUtil;
+import org.apache.pulsar.common.util.netty.EventLoopUtil;
+import org.awaitility.Awaitility;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker-impl")
+public class SameAuthParamsLookupAutoClusterFailoverTest {
+
+ private static final String URL0 = "pulsar://broker0:6650";
+ private static final String URL1 = "pulsar://broker1:6650";
+ private static final String URL2 = "pulsar://broker2:6650";
+
+ private EventLoopGroup executor;
+ private PulsarClientImpl mockClient;
+ private SameAuthParamsLookupAutoClusterFailover failover;
+ private PulsarServiceState[] stateArray;
+ private MutableInt[] counterArray;
+
+ @BeforeMethod
+ public void setup() throws Exception {
+ executor = EventLoopUtil.newEventLoopGroup(1, false,
+ new ExecutorProvider.ExtendedThreadFactory("test-failover"));
+
+ String[] urlArray = new String[]{URL0, URL1, URL2};
+ failover = SameAuthParamsLookupAutoClusterFailover.builder()
+ .pulsarServiceUrlArray(urlArray)
+ .failoverThreshold(1)
+ .recoverThreshold(2)
+ .checkHealthyIntervalMs(100)
+ .testTopic("a/b/c")
+ .markTopicNotFoundAsAvailable(true)
+ .build();
+
+ mockClient = mock(PulsarClientImpl.class);
+ doNothing().when(mockClient).updateServiceUrl(anyString());
+ doNothing().when(mockClient).reloadLookUp();
+
+ FieldUtils.writeField(failover, "pulsarClient", mockClient, true);
+ FieldUtils.writeField(failover, "executor", executor, true);
+
+ stateArray = (PulsarServiceState[]) FieldUtils.readField(failover, "pulsarServiceStateArray", true);
+ counterArray = (MutableInt[]) FieldUtils.readField(failover, "checkCounterArray", true);
+ }
+
+ @AfterMethod(alwaysRun = true)
+ public void cleanup() {
+ if (executor != null) {
+ executor.shutdownGracefully(0, 0, TimeUnit.MILLISECONDS);
+ }
+ }
+
+ private void setLookupResult(String url, boolean available) {
+ LookupService lookup = mock(LookupService.class);
+ if (available) {
+ InetSocketAddress addr = InetSocketAddress.createUnresolved("broker", 6650);
+ when(lookup.getBroker(any()))
+ .thenReturn(CompletableFuture.completedFuture(
+ new LookupTopicResult(addr, addr, false)));
+ } else {
+ when(lookup.getBroker(any()))
+ .thenReturn(FutureUtil.failedFuture(
+ new RuntimeException("connection refused")));
+ }
+ when(mockClient.getLookup(url)).thenReturn(lookup);
+ }
+
+ /**
+ * Reproduces the race condition where findFailoverTo() skips over an unavailable service
+ * without marking it as Failed. This leaves a stale Healthy state that causes a spurious
+ * recovery bounce (0 -> 2 -> 1 -> 2) instead of a clean failover (0 -> 2).
+ *
+ * <p>Before the fix, after failover from index 0 to 2, state[1] remained Healthy (stale).
+ * On the next check cycle, firstHealthyPulsarService() would see state[1]=Healthy and
+ * immediately "recover" to index 1 — which is actually a broken service. This caused
+ * unnecessary bouncing and, combined with 3-second probe timeouts on dead services,
+ * could push the total failover time past the test's awaitility timeout.
+ */
+ @Test(timeOut = 30000)
+ public void testFindFailoverToMarksSkippedServicesAsFailed() throws Exception {
+ // url0 is down, url1 is down, url2 is healthy.
+ setLookupResult(URL0, false);
+ setLookupResult(URL1, false);
+ setLookupResult(URL2, true);
+
+ // Pre-set state[0] to Failed (as if checkPulsarServices already detected it),
+ // then run one check cycle. All on the executor to ensure thread safety.
+ runOnExecutor(() -> {
+ stateArray[0] = PulsarServiceState.Failed;
+ counterArray[0].setValue(0);
+ });
+ runCheckCycle();
+
+ // After the fix, findFailoverTo marks url1 as Failed when it fails probing.
+ // Verify on the executor thread where state is owned.
+ runOnExecutor(() -> {
+ assertEquals(failover.getCurrentPulsarServiceIndex(), 2,
+ "Should have failed over to index 2");
+ assertEquals(stateArray[1], PulsarServiceState.Failed,
+ "Service 1 should be marked Failed by findFailoverTo, not remain stale Healthy");
+ assertEquals(stateArray[2], PulsarServiceState.Healthy,
+ "Service 2 should remain Healthy");
+ });
+ }
+
+ /**
+ * Verifies no spurious recovery bounce occurs after failover. Without the fix,
+ * the first check cycle after failover to index 2 would see stale Healthy state[1]
+ * and immediately switch to index 1.
+ */
+ @Test(timeOut = 30000)
+ public void testNoSpuriousRecoveryBounceAfterFailover() throws Exception {
+ // url0 is down, url1 is down, url2 is healthy.
+ setLookupResult(URL0, false);
+ setLookupResult(URL1, false);
+ setLookupResult(URL2, true);
+
+ // Pre-set state[0] to Failed.
+ runOnExecutor(() -> {
+ stateArray[0] = PulsarServiceState.Failed;
+ counterArray[0].setValue(0);
+ });
+
+ // Failover: 0 -> 2.
+ runCheckCycle();
+ runOnExecutor(() -> assertEquals(failover.getCurrentPulsarServiceIndex(), 2));
+
+ // Run another check cycle. Without the fix, state[1] would be stale Healthy,
+ // and firstHealthyPulsarService would return 1, causing a spurious switch.
+ runCheckCycle();
+ runOnExecutor(() -> assertEquals(failover.getCurrentPulsarServiceIndex(), 2,
+ "Should stay at index 2, not bounce to index 1"));
+ }
+
+ /**
+ * Verifies that recovery still works correctly for a service that was marked Failed
+ * by findFailoverTo, once that service becomes available again.
+ */
+ @Test(timeOut = 30000)
+ public void testRecoveryAfterFindFailoverToMarksServiceFailed() throws Exception {
+ // url0 is down, url1 is down, url2 is healthy.
+ setLookupResult(URL0, false);
+ setLookupResult(URL1, false);
+ setLookupResult(URL2, true);
+
+ // Pre-set state[0] to Failed and trigger failover 0 -> 2.
+ runOnExecutor(() -> {
+ stateArray[0] = PulsarServiceState.Failed;
+ counterArray[0].setValue(0);
+ });
+ runCheckCycle();
+ runOnExecutor(() -> {
+ assertEquals(failover.getCurrentPulsarServiceIndex(), 2);
+ assertEquals(stateArray[1], PulsarServiceState.Failed);
+ });
+
+ // Now make url1 healthy (simulating recovery of that service).
+ setLookupResult(URL1, true);
+
+ // Run check cycles until service 1 recovers.
+ // Failed -> PreRecover (1 check) -> Healthy (recoverThreshold=2, so 1 more check).
+ Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> {
+ runCheckCycle();
+ runOnExecutor(() -> {
+ assertEquals(failover.getCurrentPulsarServiceIndex(), 1,
+ "Should recover to index 1 after it becomes healthy");
+ assertEquals(stateArray[1], PulsarServiceState.Healthy);
+ });
+ });
+ }
+
+ private void runOnExecutor(Runnable task) throws Exception {
+ try {
+ executor.submit(task).get(5, TimeUnit.SECONDS);
+ } catch (java.util.concurrent.ExecutionException e) {
+ // Unwrap so that AssertionErrors propagate directly to Awaitility.
+ if (e.getCause() instanceof AssertionError) {
+ throw (AssertionError) e.getCause();
+ }
+ throw e;
+ }
+ }
+
+ private void runCheckCycle() throws Exception {
+ runOnExecutor(() -> {
+ try {
+ Method checkMethod = SameAuthParamsLookupAutoClusterFailover.class
+ .getDeclaredMethod("checkPulsarServices");
+ checkMethod.setAccessible(true);
+ Method firstHealthyMethod = SameAuthParamsLookupAutoClusterFailover.class
+ .getDeclaredMethod("firstHealthyPulsarService");
+ firstHealthyMethod.setAccessible(true);
+ Method findFailoverMethod = SameAuthParamsLookupAutoClusterFailover.class
+ .getDeclaredMethod("findFailoverTo");
+ findFailoverMethod.setAccessible(true);
+ Method updateMethod = SameAuthParamsLookupAutoClusterFailover.class
+ .getDeclaredMethod("updateServiceUrl", int.class);
+ updateMethod.setAccessible(true);
+
+ checkMethod.invoke(failover);
+ int firstHealthy = (int) firstHealthyMethod.invoke(failover);
+ int currentIndex = failover.getCurrentPulsarServiceIndex();
+ if (firstHealthy != currentIndex) {
+ if (firstHealthy < 0) {
+ int failoverTo = (int) findFailoverMethod.invoke(failover);
+ if (failoverTo >= 0) {
+ updateMethod.invoke(failover, failoverTo);
+ }
+ } else {
+ updateMethod.invoke(failover, firstHealthy);
+ }
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+ }
+}
(pulsar) 03/05: [fix][broker] Fix stuck chunks in SharedConsumerAssignor permit tracking (#25620)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 0693715ca096e55cb52e294decdbf39496d3ecec
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Thu Apr 30 06:18:54 2026 -0700
[fix][broker] Fix stuck chunks in SharedConsumerAssignor permit tracking (#25620)
(cherry picked from commit 759a5f520f81ba45caef9bd94a79997855a695fd)
---
.../apache/pulsar/broker/service/SharedConsumerAssignor.java | 10 +++++-----
1 file changed, 5 insertions(+), 5 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SharedConsumerAssignor.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SharedConsumerAssignor.java
index bbf8dfd2b10..a317ad7560b 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SharedConsumerAssignor.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SharedConsumerAssignor.java
@@ -89,7 +89,7 @@ public class SharedConsumerAssignor {
if (metadata == null || !metadata.hasUuid() || !metadata.hasChunkId() || !metadata.hasNumChunksFromMsg()) {
consumerToEntries.computeIfAbsent(consumer, __ -> new ArrayList<>()).add(entryAndMetadata);
} else {
- final Consumer consumerForUuid = getConsumerForUuid(metadata, consumer, availablePermits);
+ final Consumer consumerForUuid = getConsumerForUuid(metadata, consumer);
if (consumerForUuid == null) {
unassignedMessageProcessor.accept(entryAndMetadata);
continue;
@@ -120,9 +120,7 @@ public class SharedConsumerAssignor {
return null;
}
- private Consumer getConsumerForUuid(final MessageMetadata metadata,
- final Consumer defaultConsumer,
- final int currentAvailablePermits) {
+ private Consumer getConsumerForUuid(final MessageMetadata metadata, final Consumer defaultConsumer) {
final String uuid = metadata.getUuid();
Consumer consumer = uuidToConsumer.get(uuid);
if (consumer == null) {
@@ -141,7 +139,9 @@ public class SharedConsumerAssignor {
// The last chunk is received, we should remove the cache
uuidToConsumer.remove(uuid);
}
- consumerToPermits.put(consumer, currentAvailablePermits - 1);
+ // Decrement target consumer's permits, not the loop's local availablePermits — on a cache
+ // redirect those track different consumers.
+ consumerToPermits.put(consumer, permits - 1);
return consumer;
}
}
(pulsar) 02/05: [fix][broker] Fix precision loss in DataSketchesSummaryLogger by replacing LongAdder with DoubleAdder for sum accumulation (#25594)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 644b55bee983d16c7950f5ad4c8b28a9c5d99645
Author: zhenJiangWang <zh...@gmail.com>
AuthorDate: Wed Apr 29 10:47:18 2026 +0800
[fix][broker] Fix precision loss in DataSketchesSummaryLogger by replacing LongAdder with DoubleAdder for sum accumulation (#25594)
(cherry picked from commit 00577a5438bbb3d959e5ca2ea037e760219b95d8)
---
.../broker/stats/prometheus/metrics/DataSketchesSummaryLogger.java | 7 ++++---
1 file changed, 4 insertions(+), 3 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/DataSketchesSummaryLogger.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/DataSketchesSummaryLogger.java
index 42c189d4bf3..7495f057aa0 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/DataSketchesSummaryLogger.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/DataSketchesSummaryLogger.java
@@ -22,6 +22,7 @@ import com.yahoo.sketches.quantiles.DoublesSketch;
import com.yahoo.sketches.quantiles.DoublesUnion;
import com.yahoo.sketches.quantiles.DoublesUnionBuilder;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.DoubleAdder;
import java.util.concurrent.atomic.LongAdder;
public class DataSketchesSummaryLogger {
@@ -37,7 +38,7 @@ public class DataSketchesSummaryLogger {
*/
private volatile DoublesSketch values;
private final LongAdder countAdder = new LongAdder();
- private final LongAdder sumAdder = new LongAdder();
+ private final DoubleAdder sumAdder = new DoubleAdder();
public DataSketchesSummaryLogger() {
this.current = new ThreadLocalAccessor();
@@ -48,7 +49,7 @@ public class DataSketchesSummaryLogger {
double valueMillis = unit.toMicros(eventLatency) / 1000.0;
countAdder.increment();
- sumAdder.add((long) valueMillis);
+ sumAdder.add(valueMillis);
current.getLocalData().updateSuccess(valueMillis);
}
@@ -69,7 +70,7 @@ public class DataSketchesSummaryLogger {
return countAdder.sum();
}
- public long getSum() {
+ public double getSum() {
return sumAdder.sum();
}
(pulsar) 01/05: [improve][client] Best-effort retry for individual/batch-index acks on send failure when ackReceiptEnabled=false (#25525)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit bb0b7d5a7762dadacb312408be3fc5c5c534b1ac
Author: Zixuan Liu <no...@gmail.com>
AuthorDate: Fri Apr 24 10:53:23 2026 +0800
[improve][client] Best-effort retry for individual/batch-index acks on send failure when ackReceiptEnabled=false (#25525)
Co-authored-by: Copilot <22...@users.noreply.github.com>
(cherry picked from commit fae3df958f74f2afb0ba398b69d94fd3e6358c4f)
---
.../org/apache/pulsar/client/impl/ClientCnx.java | 26 +++---
.../PersistentAcknowledgmentsGroupingTracker.java | 75 ++++++++++++++---
.../impl/AcknowledgementsGroupingTrackerTest.java | 94 +++++++++++++++++++++-
.../impl/ClientCnxRequestTimeoutQueueTest.java | 1 +
.../pulsar/client/impl/ClientTestFixtures.java | 5 ++
5 files changed, 174 insertions(+), 27 deletions(-)
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java
index 35c49fb46a8..39822263906 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java
@@ -1053,27 +1053,23 @@ public class ClientCnx extends PulsarHandler {
}
private <T> void sendRequestAndHandleTimeout(ByteBuf requestMessage, long requestId,
- RequestType requestType, boolean flush,
- TimedCompletableFuture<T> future) {
+ RequestType requestType, boolean flush,
+ TimedCompletableFuture<T> future) {
pendingRequests.put(requestId, future);
- if (flush) {
- ctx.writeAndFlush(requestMessage).addListener(writeFuture -> {
- if (!writeFuture.isSuccess()) {
- if (pendingRequests.remove(requestId, future) && !future.isDone()) {
- log.warn("{} Failed to send {} to broker: {}", ctx.channel(),
- requestType.getDescription(), writeFuture.cause().getMessage());
- future.completeExceptionally(writeFuture.cause());
- }
+ (flush ? ctx.writeAndFlush(requestMessage) : ctx.write(requestMessage)).addListener(writeFuture -> {
+ if (!writeFuture.isSuccess()) {
+ if (pendingRequests.remove(requestId, future) && !future.isDone()) {
+ log.warn("{} Failed to send {} to broker: {}", ctx.channel(),
+ requestType.getDescription(), writeFuture.cause().getMessage());
+ future.completeExceptionally(writeFuture.cause());
}
- });
- } else {
- ctx.write(requestMessage, ctx().voidPromise());
- }
+ }
+ });
requestTimeoutQueue.add(new RequestTime(requestId, requestType));
}
private <T> CompletableFuture<T> sendRequestAndHandleTimeout(ByteBuf requestMessage, long requestId,
- RequestType requestType, boolean flush) {
+ RequestType requestType, boolean flush) {
TimedCompletableFuture<T> future = new TimedCompletableFuture<>();
sendRequestAndHandleTimeout(requestMessage, requestId, requestType, flush, future);
return future;
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java
index 0598dc4fb36..0a366a759d3 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java
@@ -26,6 +26,7 @@ import io.netty.util.concurrent.FastThreadLocal;
import java.util.ArrayList;
import java.util.BitSet;
import java.util.Collections;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
@@ -279,6 +280,11 @@ public class PersistentAcknowledgmentsGroupingTracker implements Acknowledgments
return CompletableFuture.completedFuture(null);
}
+ @VisibleForTesting
+ int getPendingIndividualAcksSize() {
+ return pendingIndividualAcks.size();
+ }
+
private CompletableFuture<Void> doIndividualBatchAck(MessageIdAdv batchMessageId,
Map<String, Long> properties) {
if (acknowledgementGroupTimeMicros == 0 || (properties != null && !properties.isEmpty())) {
@@ -405,7 +411,7 @@ public class PersistentAcknowledgmentsGroupingTracker implements Acknowledgments
}
CompletableFuture<Void> completableFuture = newMessageAckCommandAndWrite(cnx, consumer.consumerId,
- msgId.getLedgerId(), msgId.getEntryId(), bitSet, ackType, properties, true, null, null);
+ msgId.getLedgerId(), msgId.getEntryId(), bitSet, ackType, properties, true, null, null, null);
bitSet.recycle();
return completableFuture;
}
@@ -441,13 +447,15 @@ public class PersistentAcknowledgmentsGroupingTracker implements Acknowledgments
newMessageAckCommandAndWrite(cnx, consumer.consumerId, messageId.getLedgerId(), messageId.getEntryId(),
lastCumulativeAckToFlush.getBitSetRecyclable(), AckType.Cumulative,
Collections.emptyMap(), false,
- (TimedCompletableFuture<Void>) this.currentCumulativeAckFuture, null);
+ (TimedCompletableFuture<Void>) this.currentCumulativeAckFuture, null, null);
this.consumer.unAckedChunkedMessageIdSequenceMap.remove(messageId);
}
// Flush all individual acks
List<Triple<Long, Long, ConcurrentBitSet>> entriesToAck =
new ArrayList<>(pendingIndividualAcks.size() + pendingIndividualBatchIndexAcks.size());
+ List<MessageIdAdv> individualAcksToFlush = new ArrayList<>(pendingIndividualAcks.size());
+ Map<MessageIdAdv, MessageIdImpl[]> chunkedMessageIdsToRestore = new HashMap<>();
if (!pendingIndividualAcks.isEmpty()) {
if (Commands.peerSupportsMultiMessageAcknowledgment(cnx.getRemoteEndpointProtocolVersion())) {
// We can send 1 single protobuf command with all individual acks
@@ -456,11 +464,13 @@ public class PersistentAcknowledgmentsGroupingTracker implements Acknowledgments
if (msgId == null) {
break;
}
+ individualAcksToFlush.add(msgId);
// if messageId is checked then all the chunked related to that msg also processed so, ack all of
// them
MessageIdImpl[] chunkMsgIds = this.consumer.unAckedChunkedMessageIdSequenceMap.get(msgId);
if (chunkMsgIds != null && chunkMsgIds.length > 1) {
+ chunkedMessageIdsToRestore.put(msgId, chunkMsgIds);
for (MessageIdImpl cMsgId : chunkMsgIds) {
if (cMsgId != null) {
entriesToAck.add(Triple.of(cMsgId.getLedgerId(), cMsgId.getEntryId(), null));
@@ -479,14 +489,16 @@ public class PersistentAcknowledgmentsGroupingTracker implements Acknowledgments
if (msgId == null) {
break;
}
+ individualAcksToFlush.add(msgId);
newMessageAckCommandAndWrite(cnx, consumer.consumerId, msgId.getLedgerId(), msgId.getEntryId(),
null, AckType.Individual, Collections.emptyMap(), false,
- null, null);
+ null, null, () -> restoreIndividualAck(msgId, null));
shouldFlush = true;
}
}
}
+ List<Map.Entry<MessageIdAdv, ConcurrentBitSet>> batchIndexAcksToFlush = new ArrayList<>();
while (true) {
Map.Entry<MessageIdAdv, ConcurrentBitSet> entry =
pendingIndividualBatchIndexAcks.pollFirstEntry();
@@ -494,6 +506,7 @@ public class PersistentAcknowledgmentsGroupingTracker implements Acknowledgments
// The entry has been removed in a different thread
break;
}
+ batchIndexAcksToFlush.add(entry);
entriesToAck.add(Triple.of(
entry.getKey().getLedgerId(), entry.getKey().getEntryId(), entry.getValue()));
}
@@ -502,7 +515,9 @@ public class PersistentAcknowledgmentsGroupingTracker implements Acknowledgments
newMessageAckCommandAndWrite(cnx, consumer.consumerId, 0L, 0L,
null, AckType.Individual, null, true,
- (TimedCompletableFuture<Void>) currentIndividualAckFuture, entriesToAck);
+ (TimedCompletableFuture<Void>) currentIndividualAckFuture, entriesToAck,
+ () -> restoreIndividualAndBatchIndexAcks(individualAcksToFlush, chunkedMessageIdsToRestore,
+ batchIndexAcksToFlush));
shouldFlush = true;
}
@@ -547,19 +562,19 @@ public class PersistentAcknowledgmentsGroupingTracker implements Acknowledgments
}
}
completableFuture = newMessageAckCommandAndWrite(cnx, consumer.consumerId, 0L, 0L,
- null, ackType, null, true, null, entriesToAck);
+ null, ackType, null, true, null, entriesToAck, null);
} else {
// if don't support multi message ack, it also support ack receipt, so we should not think about the
// ack receipt in this logic
for (MessageIdImpl cMsgId : chunkMsgIds) {
newMessageAckCommandAndWrite(cnx, consumerId, cMsgId.getLedgerId(), cMsgId.getEntryId(),
- bitSet, ackType, map, true, null, null);
+ bitSet, ackType, map, true, null, null, null);
}
completableFuture = CompletableFuture.completedFuture(null);
}
} else {
completableFuture = newMessageAckCommandAndWrite(cnx, consumerId, msgId.getLedgerId(), msgId.getEntryId(),
- bitSet, ackType, map, true, null, null);
+ bitSet, ackType, map, true, null, null, null);
}
return completableFuture;
}
@@ -569,7 +584,8 @@ public class PersistentAcknowledgmentsGroupingTracker implements Acknowledgments
long entryId, BitSetRecyclable ackSet, AckType ackType,
Map<String, Long> properties, boolean flush,
TimedCompletableFuture<Void> timedCompletableFuture,
- List<Triple<Long, Long, ConcurrentBitSet>> entriesToAck) {
+ List<Triple<Long, Long, ConcurrentBitSet>> entriesToAck,
+ Runnable writeFailureCallback) {
if (consumer.isAckReceiptEnabled()) {
final long requestId = consumer.getClient().newRequestId();
final ByteBuf cmd;
@@ -611,14 +627,53 @@ public class PersistentAcknowledgmentsGroupingTracker implements Acknowledgments
cmd = Commands.newMultiMessageAck(consumerId, entriesToAck, -1);
}
if (flush) {
- cnx.ctx().writeAndFlush(cmd, cnx.ctx().voidPromise());
+ if (writeFailureCallback == null) {
+ cnx.ctx().writeAndFlush(cmd, cnx.ctx().voidPromise());
+ } else {
+ cnx.ctx().writeAndFlush(cmd).addListener(writeFuture -> {
+ if (!writeFuture.isSuccess()) {
+ writeFailureCallback.run();
+ }
+ });
+ }
} else {
- cnx.ctx().write(cmd, cnx.ctx().voidPromise());
+ if (writeFailureCallback == null) {
+ cnx.ctx().write(cmd, cnx.ctx().voidPromise());
+ } else {
+ cnx.ctx().write(cmd).addListener(writeFuture -> {
+ if (!writeFuture.isSuccess()) {
+ writeFailureCallback.run();
+ }
+ });
+ }
}
return CompletableFuture.completedFuture(null);
}
}
+ private void restoreIndividualAck(MessageIdAdv messageId, @Nullable MessageIdImpl[] chunkMsgIds) {
+ pendingIndividualAcks.add(messageId);
+ restoreChunkedMessageIds(messageId, chunkMsgIds);
+ }
+
+ private void restoreIndividualAndBatchIndexAcks(List<MessageIdAdv> messageIds,
+ Map<MessageIdAdv, MessageIdImpl[]> chunkMsgIds,
+ List<Map.Entry<MessageIdAdv, ConcurrentBitSet>> batchIndexAcks) {
+ pendingIndividualAcks.addAll(messageIds);
+ chunkMsgIds.forEach(this::restoreChunkedMessageIds);
+ batchIndexAcks.forEach(entry -> pendingIndividualBatchIndexAcks.merge(entry.getKey(), entry.getValue(),
+ (currentValue, valueToRestore) -> {
+ currentValue.and(valueToRestore);
+ return currentValue;
+ }));
+ }
+
+ private void restoreChunkedMessageIds(MessageIdAdv messageId, @Nullable MessageIdImpl[] chunkMsgIds) {
+ if (chunkMsgIds != null) {
+ consumer.unAckedChunkedMessageIdSequenceMap.putIfAbsent(messageId, chunkMsgIds);
+ }
+ }
+
public Optional<Lock> acquireReadLock() {
Optional<Lock> optionalLock = Optional.ofNullable(consumer.isAckReceiptEnabled() ? lock.readLock() : null);
optionalLock.ifPresent(Lock::lock);
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/AcknowledgementsGroupingTrackerTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/AcknowledgementsGroupingTrackerTest.java
index 76f2f45a030..410819ef732 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/AcknowledgementsGroupingTrackerTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/AcknowledgementsGroupingTrackerTest.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.client.impl;
+import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
@@ -28,9 +29,13 @@ import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertTrue;
import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.util.ReferenceCountUtil;
+import io.netty.util.concurrent.Future;
+import io.netty.util.concurrent.GenericFutureListener;
import java.util.ArrayList;
import java.util.BitSet;
import java.util.Collections;
@@ -46,6 +51,7 @@ import org.apache.pulsar.client.impl.metrics.InstrumentProvider;
import org.apache.pulsar.client.util.TimedCompletableFuture;
import org.apache.pulsar.common.api.proto.CommandAck.AckType;
import org.apache.pulsar.common.api.proto.ProtocolVersion;
+import org.apache.pulsar.common.util.FutureUtil;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.DataProvider;
@@ -57,6 +63,8 @@ public class AcknowledgementsGroupingTrackerTest {
private ConsumerImpl<?> consumer;
private EventLoopGroup eventLoopGroup;
private AtomicBoolean returnCnx = new AtomicBoolean(true);
+ private ChannelHandlerContext successCtx;
+ private AtomicBoolean failAckCommandSend = new AtomicBoolean(false);
@BeforeClass
public void setup() throws NoSuchFieldException, IllegalAccessException {
@@ -71,9 +79,9 @@ public class AcknowledgementsGroupingTrackerTest {
doReturn(new ConsumerStatsRecorderImpl()).when(consumer).getStats();
doReturn(UnAckedMessageTracker.UNACKED_MESSAGE_TRACKER_DISABLED)
.when(consumer).getUnAckedMessageTracker();
- ChannelHandlerContext ctx = ClientTestFixtures.mockChannelHandlerContext();
+ successCtx = ClientTestFixtures.mockChannelHandlerContext();
doAnswer(invocation -> returnCnx.get() ? cnx : null).when(consumer).getClientCnx();
- doReturn(ctx).when(cnx).ctx();
+ doReturn(successCtx).when(cnx).ctx();
}
@DataProvider(name = "isNeedReceipt")
@@ -324,6 +332,60 @@ public class AcknowledgementsGroupingTrackerTest {
tracker.close();
}
+ @Test
+ public void testFlushRetainsPendingIndividualAckOnSendFailureWithoutAckReceipt() throws Exception {
+ ConsumerConfigurationData<?> conf = new ConsumerConfigurationData<>();
+ conf.setAcknowledgementsGroupTimeMicros(TimeUnit.SECONDS.toMicros(10));
+ conf.setAckReceiptEnabled(false);
+ doReturn(false).when(consumer).isAckReceiptEnabled();
+ PersistentAcknowledgmentsGroupingTracker tracker =
+ new PersistentAcknowledgmentsGroupingTracker(consumer, conf, eventLoopGroup);
+
+ MessageIdImpl msg1 = new MessageIdImpl(5, 1, 0);
+ tracker.addAcknowledgment(msg1, AckType.Individual, Collections.emptyMap());
+ assertEquals(tracker.getPendingIndividualAcksSize(), 1);
+
+ doReturn(createFailedChannelHandlerContext()).when(cnx).ctx();
+
+ tracker.flush();
+
+ assertTrue(tracker.isDuplicate(msg1));
+ assertEquals(tracker.getPendingIndividualAcksSize(), 1);
+
+ doReturn(successCtx).when(cnx).ctx();
+
+ tracker.flush();
+
+ assertFalse(tracker.isDuplicate(msg1));
+ assertEquals(tracker.getPendingIndividualAcksSize(), 0);
+ tracker.close();
+ }
+
+ @Test
+ public void testFlushFailsAckFutureOnSendFailureWithAckReceipt() throws Exception {
+ ConsumerConfigurationData<?> conf = new ConsumerConfigurationData<>();
+ conf.setAcknowledgementsGroupTimeMicros(TimeUnit.SECONDS.toMicros(10));
+ conf.setAckReceiptEnabled(true);
+ doReturn(true).when(consumer).isAckReceiptEnabled();
+ PersistentAcknowledgmentsGroupingTracker tracker =
+ new PersistentAcknowledgmentsGroupingTracker(consumer, conf, eventLoopGroup);
+
+ MessageIdImpl msg1 = new MessageIdImpl(5, 1, 0);
+ CompletableFuture<Void> ackFuture =
+ tracker.addAcknowledgment(msg1, AckType.Individual, Collections.emptyMap());
+ assertEquals(tracker.getPendingIndividualAcksSize(), 1);
+
+ failAckCommandSend.set(true);
+ tracker.flush();
+
+ assertTrue(ackFuture.isCompletedExceptionally());
+ assertFalse(tracker.isDuplicate(msg1));
+ assertEquals(tracker.getPendingIndividualAcksSize(), 0);
+
+ failAckCommandSend.set(false);
+ tracker.close();
+ }
+
@Test(dataProvider = "isNeedReceipt")
public void testBatchAckTrackerMultiAck(boolean isNeedReceipt) throws Exception {
ConsumerConfigurationData<?> conf = new ConsumerConfigurationData<>();
@@ -464,12 +526,40 @@ public class AcknowledgementsGroupingTrackerTest {
@Override
public CompletableFuture<Void> newAckForReceipt(ByteBuf request, long requestId) {
+ if (failAckCommandSend.get()) {
+ return FutureUtil.failedFuture(new RuntimeException("ack send failed"));
+ }
return CompletableFuture.completedFuture(null);
}
@Override
public void newAckForReceiptWithFuture(ByteBuf request, long requestId,
TimedCompletableFuture<Void> future) {
+ if (failAckCommandSend.get()) {
+ future.completeExceptionally(new RuntimeException("ack send failed"));
+ }
}
}
+
+ private ChannelHandlerContext createFailedChannelHandlerContext() {
+ ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
+ ChannelFuture listenerFuture = mock(ChannelFuture.class);
+ ChannelFuture failedFuture = mock(ChannelFuture.class);
+ when(failedFuture.isSuccess()).thenReturn(false);
+ when(failedFuture.cause()).thenReturn(new RuntimeException("ack send failed"));
+ doAnswer(invocation -> {
+ GenericFutureListener<Future<Void>> listener = invocation.getArgument(0);
+ listener.operationComplete(failedFuture);
+ return listenerFuture;
+ }).when(listenerFuture).addListener(any());
+ doAnswer(invocation -> {
+ ReferenceCountUtil.release(invocation.getArgument(0));
+ return listenerFuture;
+ }).when(ctx).write(any());
+ doAnswer(invocation -> {
+ ReferenceCountUtil.release(invocation.getArgument(0));
+ return listenerFuture;
+ }).when(ctx).writeAndFlush(any());
+ return ctx;
+ }
}
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientCnxRequestTimeoutQueueTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientCnxRequestTimeoutQueueTest.java
index c4dab1ad351..7e5d98b136a 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientCnxRequestTimeoutQueueTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientCnxRequestTimeoutQueueTest.java
@@ -64,6 +64,7 @@ public class ClientCnxRequestTimeoutQueueTest {
ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
Channel channel = mock(Channel.class);
when(ctx.writeAndFlush(any())).thenAnswer(args -> mock(ChannelFuture.class));
+ when(ctx.write(any())).thenAnswer(args -> mock(ChannelFuture.class));
when(ctx.channel()).thenReturn(channel);
when(channel.remoteAddress()).thenReturn(new InetSocketAddress(1234));
cnx.channelActive(ctx);
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientTestFixtures.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientTestFixtures.java
index ae0797fa493..494dea365d9 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientTestFixtures.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientTestFixtures.java
@@ -147,6 +147,11 @@ class ClientTestFixtures {
}).when(listenerFuture).addListener(any());
// handle write and writeAndFlush methods so that the input message is released
+ doAnswer(invocation -> {
+ Object msg = invocation.getArgument(0);
+ ReferenceCountUtil.release(msg);
+ return listenerFuture;
+ }).when(ctx).write(any());
doAnswer(invocation -> {
Object msg = invocation.getArgument(0);
ReferenceCountUtil.release(msg);
(pulsar) 05/05: [fix][client] Reset higher-index states on recovery in SameAuthParamsLookupAutoClusterFailover (#25826)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 182d8858bd95a047d876793206f2c0cb50b59c37
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Wed May 20 06:48:58 2026 -0700
[fix][client] Reset higher-index states on recovery in SameAuthParamsLookupAutoClusterFailover (#25826)
(cherry picked from commit 4229e197479e8ea10164c30abf95c41ceff2b344)
---
.../SameAuthParamsLookupAutoClusterFailover.java | 11 ++++
...ameAuthParamsLookupAutoClusterFailoverTest.java | 63 ++++++++++++++++++++++
2 files changed, 74 insertions(+)
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/SameAuthParamsLookupAutoClusterFailover.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/SameAuthParamsLookupAutoClusterFailover.java
index 5ae7b96906e..448dd95c4fd 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/SameAuthParamsLookupAutoClusterFailover.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/SameAuthParamsLookupAutoClusterFailover.java
@@ -261,6 +261,17 @@ public class SameAuthParamsLookupAutoClusterFailover implements ServiceUrlProvid
try {
pulsarClient.updateServiceUrl(targetUrl);
pulsarClient.reloadLookUp();
+ // When recovering to a higher-priority service, the check loop will only probe
+ // indices 0..targetIndex going forward. Any transient state (e.g., PreFail from
+ // a single timed-out probe) at higher indices would become stuck because those
+ // indices are no longer probed. Reset them so they start fresh if a future
+ // failover needs to consider them again.
+ if (targetIndex < currentPulsarServiceIndex) {
+ for (int i = targetIndex + 1; i < pulsarServiceStateArray.length; i++) {
+ pulsarServiceStateArray[i] = PulsarServiceState.Healthy;
+ checkCounterArray[i].setValue(0);
+ }
+ }
currentPulsarServiceIndex = targetIndex;
} catch (Exception e) {
log.error("Failed to {}", logMsg, e);
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/SameAuthParamsLookupAutoClusterFailoverTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/SameAuthParamsLookupAutoClusterFailoverTest.java
index 31e2bf6aaed..959d913dff8 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/SameAuthParamsLookupAutoClusterFailoverTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/SameAuthParamsLookupAutoClusterFailoverTest.java
@@ -168,6 +168,69 @@ public class SameAuthParamsLookupAutoClusterFailoverTest {
"Should stay at index 2, not bounce to index 1"));
}
+ /**
+ * Reproduces the bug fixed by resetting state of higher-priority-than-target indices on
+ * recovery. The check loop only probes indices 0..currentPulsarServiceIndex, so if a
+ * higher index was left in a transient state (e.g., PreFail from a single timed-out
+ * probe) at the moment of recovery, it would stay there forever because no future probe
+ * ever visits it.
+ *
+ * <p>Scenario: failover 0 -> 2 (state[2]=Healthy). url1 recovers and is about to trigger
+ * recovery 2 -> 1. On the same check cycle that promotes state[1] to Healthy, url2 sees
+ * one transient probe failure that flips state[2] from Healthy to PreFail. The recovery
+ * fires (current 2 -> 1) and from that point on the loop only probes indices 0 and 1.
+ *
+ * <p>Without the fix, state[2] stays at PreFail forever. With the fix, state[2] is
+ * reset to Healthy when the recovery transition runs.
+ */
+ @Test(timeOut = 30000)
+ public void testRecoveryResetsHigherIndexStaleState() throws Exception {
+ // url0 down, url1 down, url2 up.
+ setLookupResult(URL0, false);
+ setLookupResult(URL1, false);
+ setLookupResult(URL2, true);
+
+ // Pre-set state[0]=Failed and trigger failover 0 -> 2.
+ runOnExecutor(() -> {
+ stateArray[0] = PulsarServiceState.Failed;
+ counterArray[0].setValue(0);
+ });
+ runCheckCycle();
+ runOnExecutor(() -> {
+ assertEquals(failover.getCurrentPulsarServiceIndex(), 2);
+ assertEquals(stateArray[1], PulsarServiceState.Failed);
+ assertEquals(stateArray[2], PulsarServiceState.Healthy);
+ });
+
+ // url1 becomes healthy; first check cycle moves state[1] Failed -> PreRecover.
+ setLookupResult(URL1, true);
+ runCheckCycle();
+ runOnExecutor(() -> {
+ assertEquals(failover.getCurrentPulsarServiceIndex(), 2);
+ assertEquals(stateArray[1], PulsarServiceState.PreRecover);
+ assertEquals(stateArray[2], PulsarServiceState.Healthy);
+ });
+
+ // On the cycle that completes recovery (state[1] PreRecover -> Healthy and triggers
+ // updateServiceUrl(1)), inject a single failed probe at url2 so state[2] flips
+ // Healthy -> PreFail just before the index transition.
+ setLookupResult(URL2, false);
+ runCheckCycle();
+
+ // After recovery to index 1:
+ // - With the fix: state[2] is reset to Healthy on the transition.
+ // - Without the fix: state[2] is stuck at PreFail forever — the check loop now
+ // only iterates 0..1 and never visits index 2 again.
+ runOnExecutor(() -> {
+ assertEquals(failover.getCurrentPulsarServiceIndex(), 1);
+ assertEquals(stateArray[1], PulsarServiceState.Healthy);
+ assertEquals(stateArray[2], PulsarServiceState.Healthy,
+ "state[2] should be reset to Healthy on recovery, not stuck at PreFail");
+ assertEquals(counterArray[2].intValue(), 0,
+ "counter[2] should be reset to 0 on recovery");
+ });
+ }
+
/**
* Verifies that recovery still works correctly for a service that was marked Failed
* by findFailoverTo, once that service becomes available again.