This is an automated email from the ASF dual-hosted git repository.
ffang pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/cxf.git
The following commit(s) were added to refs/heads/main by this push:
new 37b2d8147e7 [CXF-9213]make RetryStrategy stateless (#3143)
37b2d8147e7 is described below
commit 37b2d8147e7bdf0b2c61db05fc232f9b251948e0
Author: Freeman(Yue) Fang <[email protected]>
AuthorDate: Tue Jun 2 20:43:03 2026 -0400
[CXF-9213]make RetryStrategy stateless (#3143)
---
.../cxf/clustering/FailoverTargetSelector.java | 41 +++--
.../clustering/PerInvocationFailoverStrategy.java | 39 +++++
.../org/apache/cxf/clustering/RetryStrategy.java | 60 +++++--
.../apache/cxf/clustering/RetryStrategyTest.java | 181 +++++++++++++++++++++
.../jaxrs/failover/AbstractFailoverTest.java | 36 +++-
5 files changed, 328 insertions(+), 29 deletions(-)
diff --git
a/rt/features/clustering/src/main/java/org/apache/cxf/clustering/FailoverTargetSelector.java
b/rt/features/clustering/src/main/java/org/apache/cxf/clustering/FailoverTargetSelector.java
index debfd2dfab2..666e576d694 100644
---
a/rt/features/clustering/src/main/java/org/apache/cxf/clustering/FailoverTargetSelector.java
+++
b/rt/features/clustering/src/main/java/org/apache/cxf/clustering/FailoverTargetSelector.java
@@ -53,6 +53,8 @@ public class FailoverTargetSelector extends
AbstractConduitSelector {
protected FailoverStrategy failoverStrategy;
private ConcurrentHashMap<String, InvocationContext> inProgress = new
ConcurrentHashMap<>();
+ // CXF-9213: per-invocation strategy instances for
PerInvocationFailoverStrategy implementations.
+ private ConcurrentHashMap<String, FailoverStrategy> inProgressStrategies =
new ConcurrentHashMap<>();
private boolean supportNotAvailableErrorsOnly = true;
private String clientBootstrapAddress;
@@ -113,6 +115,11 @@ public class FailoverTargetSelector extends
AbstractConduitSelector {
params,
context);
inProgress.putIfAbsent(key, invocation);
+ // CXF-9213: create a fresh per-invocation instance for strategies
that carry state.
+ if (getStrategy() instanceof PerInvocationFailoverStrategy) {
+ inProgressStrategies.putIfAbsent(key,
+ ((PerInvocationFailoverStrategy)
getStrategy()).newStrategy());
+ }
}
}
@@ -179,6 +186,7 @@ public class FailoverTargetSelector extends
AbstractConduitSelector {
if (!failover) {
inProgress.remove(key);
+ inProgressStrategies.remove(key);
doComplete(exchange);
}
}
@@ -316,21 +324,19 @@ public class FailoverTargetSelector extends
AbstractConduitSelector {
*/
protected Endpoint getFailoverTarget(Exchange exchange,
InvocationContext invocation) {
+ String key = getInvocationKey(exchange);
+ FailoverStrategy strategy = getStrategy(key);
List<String> alternateAddresses = updateContextAlternatives(exchange,
invocation);
Endpoint failoverTarget = null;
if (alternateAddresses != null) {
- String alternateAddress =
- getStrategy().selectAlternateAddress(alternateAddresses);
+ String alternateAddress =
strategy.selectAlternateAddress(alternateAddresses);
if (alternateAddress != null) {
// re-use current endpoint
- //
failoverTarget = getEndpoint();
-
failoverTarget.getEndpointInfo().setAddress(alternateAddress);
}
} else {
- failoverTarget = getStrategy().selectAlternateEndpoint(
- invocation.getAlternateEndpoints());
+ failoverTarget =
strategy.selectAlternateEndpoint(invocation.getAlternateEndpoints());
}
return failoverTarget;
}
@@ -346,14 +352,12 @@ public class FailoverTargetSelector extends
AbstractConduitSelector {
final List<String> alternateAddresses;
if (!invocation.hasAlternates()) {
// no previous failover attempt on this invocation
- //
- alternateAddresses =
- getStrategy().getAlternateAddresses(exchange);
+ FailoverStrategy strategy =
getStrategy(getInvocationKey(exchange));
+ alternateAddresses = strategy.getAlternateAddresses(exchange);
if (alternateAddresses != null) {
invocation.setAlternateAddresses(alternateAddresses);
} else {
- invocation.setAlternateEndpoints(
- getStrategy().getAlternateEndpoints(exchange));
+
invocation.setAlternateEndpoints(strategy.getAlternateEndpoints(exchange));
}
} else {
alternateAddresses = invocation.getAlternateAddresses();
@@ -361,6 +365,21 @@ public class FailoverTargetSelector extends
AbstractConduitSelector {
return alternateAddresses;
}
+ /**
+ * Returns the per-invocation {@link FailoverStrategy} for the given key
if one
+ * was registered by {@link PerInvocationFailoverStrategy#newStrategy()},
otherwise
+ * returns the shared strategy.
+ */
+ protected FailoverStrategy getStrategy(String key) {
+ if (key != null) {
+ FailoverStrategy perInvocation = inProgressStrategies.get(key);
+ if (perInvocation != null) {
+ return perInvocation;
+ }
+ }
+ return getStrategy();
+ }
+
/**
* Override the ENDPOINT_ADDRESS property in the request context
*
diff --git
a/rt/features/clustering/src/main/java/org/apache/cxf/clustering/PerInvocationFailoverStrategy.java
b/rt/features/clustering/src/main/java/org/apache/cxf/clustering/PerInvocationFailoverStrategy.java
new file mode 100644
index 00000000000..78ddd950ac6
--- /dev/null
+++
b/rt/features/clustering/src/main/java/org/apache/cxf/clustering/PerInvocationFailoverStrategy.java
@@ -0,0 +1,39 @@
+/**
+ * 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.cxf.clustering;
+
+/**
+ * Marker interface for {@link FailoverStrategy} implementations that carry
+ * per-invocation mutable state (e.g. a retry counter).
+ *
+ * <p>When {@link FailoverTargetSelector} detects that the configured strategy
+ * implements this interface it calls {@link #newStrategy()} at the start of
+ * each top-level invocation and uses the returned instance for all failover
+ * decisions belonging to that invocation. The shared bean is therefore never
+ * mutated during normal processing and is safe to configure as a singleton.
+ */
+public interface PerInvocationFailoverStrategy {
+
+ /**
+ * Returns a new {@link FailoverStrategy} instance pre-configured with the
+ * same settings as {@code this} but with a fresh, zeroed mutable state.
+ * Called once per top-level invocation by {@link FailoverTargetSelector}.
+ */
+ FailoverStrategy newStrategy();
+}
diff --git
a/rt/features/clustering/src/main/java/org/apache/cxf/clustering/RetryStrategy.java
b/rt/features/clustering/src/main/java/org/apache/cxf/clustering/RetryStrategy.java
index 8e823bb7cee..ece065756ef 100644
---
a/rt/features/clustering/src/main/java/org/apache/cxf/clustering/RetryStrategy.java
+++
b/rt/features/clustering/src/main/java/org/apache/cxf/clustering/RetryStrategy.java
@@ -16,7 +16,6 @@
* specific language governing permissions and limitations
* under the License.
*/
-
package org.apache.cxf.clustering;
import java.util.List;
@@ -25,44 +24,73 @@ import org.apache.cxf.endpoint.Endpoint;
import org.apache.cxf.message.Exchange;
/**
- * This strategy simply retries the invocation using the same Endpoint
(CXF-2036).
+ * Retry strategy that retries the same endpoint for a configured number of
+ * attempts before advancing to the next alternate (CXF-2036).
+ *
+ * <p>This class implements {@link PerInvocationFailoverStrategy}: when used
+ * via {@link FailoverTargetSelector}, a fresh instance is created per
+ * top-level invocation through {@link #newStrategy()}, so the shared
+ * singleton bean is never mutated and concurrent invocations are fully
+ * isolated (CXF-9213). The per-invocation instance carries its own retry
+ * counter as a plain instance field
+ *
+ * <p>Subclasses that accumulate cross-invocation state (e.g. call counters)
+ * should override {@link #newStrategy()} to return a delegate that writes
+ * that state back to the shared instance while keeping its own retry counter.
*/
-public class RetryStrategy extends SequentialStrategy {
+public class RetryStrategy extends SequentialStrategy implements
PerInvocationFailoverStrategy {
private int maxNumberOfRetries;
- private int counter;
+ private int count;
- /* (non-Javadoc)
- * @see
org.apache.cxf.clustering.AbstractStaticFailoverStrategy#getAlternateEndpoints(
- * org.apache.cxf.message.Exchange)
- */
@Override
public List<Endpoint> getAlternateEndpoints(Exchange exchange) {
- return getEndpoints(exchange, stillTheSameAddress());
+ return getEndpoints(exchange, stillTheSameAddress(exchange));
}
@Override
protected <T> T getNextAlternate(List<T> alternates) {
- // is the amount of retries for the first alternate already exceeded?
if (!stillTheSameAddress() && !alternates.isEmpty()) {
alternates.remove(0);
}
return alternates.isEmpty() ? null : alternates.get(0);
}
+ /**
+ * Exchange-aware variant; delegates to {@link #stillTheSameAddress()} so
+ * subclasses may override either form.
+ */
+ protected boolean stillTheSameAddress(Exchange exchange) {
+ return stillTheSameAddress();
+ }
+
protected boolean stillTheSameAddress() {
if (maxNumberOfRetries == 0) {
return true;
}
- // let the target selector move to the next address
- // and then stay on the same address for maxNumberOfRetries
- if (++counter <= maxNumberOfRetries) {
+ if (++count <= maxNumberOfRetries) {
return true;
}
- counter = 0;
+ count = 0;
return false;
}
+ /**
+ * Returns a new {@link RetryStrategy} with the same configuration but a
+ * zeroed counter. Subclasses that need to accumulate state on the shared
+ * instance should override this method and return a delegating wrapper.
+ */
+ @Override
+ public FailoverStrategy newStrategy() {
+ RetryStrategy copy = new RetryStrategy();
+ copy.maxNumberOfRetries = this.maxNumberOfRetries;
+ List<String> addresses = getAlternateAddresses(null);
+ if (addresses != null) {
+ copy.setAlternateAddresses(addresses);
+ }
+ copy.setDelayBetweenRetries(getDelayBetweenRetries());
+ return copy;
+ }
public void setMaxNumberOfRetries(int maxNumberOfRetries) {
if (maxNumberOfRetries < 0) {
@@ -71,9 +99,7 @@ public class RetryStrategy extends SequentialStrategy {
this.maxNumberOfRetries = maxNumberOfRetries;
}
-
public int getMaxNumberOfRetries() {
return maxNumberOfRetries;
}
-
-}
\ No newline at end of file
+}
diff --git
a/rt/features/clustering/src/test/java/org/apache/cxf/clustering/RetryStrategyTest.java
b/rt/features/clustering/src/test/java/org/apache/cxf/clustering/RetryStrategyTest.java
new file mode 100644
index 00000000000..ad6c045e28f
--- /dev/null
+++
b/rt/features/clustering/src/test/java/org/apache/cxf/clustering/RetryStrategyTest.java
@@ -0,0 +1,181 @@
+/**
+ * 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.cxf.clustering;
+
+import java.util.Arrays;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotSame;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Unit tests for CXF-9213: RetryStrategy per-invocation isolation via
+ * {@link PerInvocationFailoverStrategy#newStrategy()}.
+ */
+public class RetryStrategyTest {
+
+ private static RetryStrategy strategyWith(int maxRetries) {
+ RetryStrategy s = new RetryStrategy();
+ s.setMaxNumberOfRetries(maxRetries);
+ return s;
+ }
+
+ // -----------------------------------------------------------------------
+ // Basic counter behaviour on a single instance
+ // -----------------------------------------------------------------------
+
+ @Test
+ public void testRetriesExactlyMaxTimes() {
+ RetryStrategy s = strategyWith(3);
+ assertTrue("attempt 1", s.stillTheSameAddress());
+ assertTrue("attempt 2", s.stillTheSameAddress());
+ assertTrue("attempt 3", s.stillTheSameAddress());
+ assertFalse("attempt 4 must exhaust", s.stillTheSameAddress());
+ }
+
+ @Test
+ public void testMaxRetriesZeroAlwaysReturnsSameAddress() {
+ RetryStrategy s = strategyWith(0);
+ for (int i = 0; i < 100; i++) {
+ assertTrue(s.stillTheSameAddress());
+ }
+ }
+
+ @Test
+ public void testCounterResetsAfterExhaustion() {
+ RetryStrategy s = strategyWith(2);
+ s.stillTheSameAddress(); // 1 – true
+ s.stillTheSameAddress(); // 2 – true
+ assertFalse(s.stillTheSameAddress()); // exhausted, resets to 0
+ assertTrue(s.stillTheSameAddress()); // new cycle starts
+ }
+
+ // -----------------------------------------------------------------------
+ // PerInvocationFailoverStrategy contract
+ // -----------------------------------------------------------------------
+
+ @Test
+ public void testImplementsPerInvocationFailoverStrategy() {
+ assertTrue(new RetryStrategy() instanceof
PerInvocationFailoverStrategy);
+ }
+
+ @Test
+ public void testNewStrategyReturnsDistinctInstance() {
+ RetryStrategy template = strategyWith(3);
+ FailoverStrategy s1 = template.newStrategy();
+ FailoverStrategy s2 = template.newStrategy();
+ assertNotSame(template, s1);
+ assertNotSame(s1, s2);
+ }
+
+ @Test
+ public void testNewStrategyInheritsMaxRetries() {
+ RetryStrategy template = strategyWith(5);
+ RetryStrategy copy = (RetryStrategy) template.newStrategy();
+ assertEquals(5, copy.getMaxNumberOfRetries());
+ }
+
+ @Test
+ public void testNewStrategyCopiesAlternateAddresses() {
+ RetryStrategy template = strategyWith(2);
+ template.setAlternateAddresses(Arrays.asList("http://a";, "http://b";));
+ RetryStrategy copy = (RetryStrategy) template.newStrategy();
+ assertEquals(Arrays.asList("http://a";, "http://b";),
copy.getAlternateAddresses(null));
+ }
+
+ @Test
+ public void testNewStrategyHasFreshCounter() {
+ RetryStrategy template = strategyWith(3);
+ // advance the template's own counter
+ template.stillTheSameAddress();
+ template.stillTheSameAddress();
+
+ // a new instance must start from zero regardless
+ RetryStrategy copy = (RetryStrategy) template.newStrategy();
+ assertTrue("copy attempt 1", copy.stillTheSameAddress());
+ assertTrue("copy attempt 2", copy.stillTheSameAddress());
+ assertTrue("copy attempt 3", copy.stillTheSameAddress());
+ assertFalse("copy attempt 4 must exhaust", copy.stillTheSameAddress());
+ }
+
+ @Test
+ public void testTwoInstancesAreIndependent() {
+ RetryStrategy template = strategyWith(3);
+ RetryStrategy i1 = (RetryStrategy) template.newStrategy();
+ RetryStrategy i2 = (RetryStrategy) template.newStrategy();
+
+ // advance i1 without exhausting it
+ i1.stillTheSameAddress();
+ i1.stillTheSameAddress();
+
+ // i2 must still be at zero
+ assertTrue("i2 attempt 1", i2.stillTheSameAddress());
+ assertTrue("i2 attempt 2", i2.stillTheSameAddress());
+ assertTrue("i2 attempt 3", i2.stillTheSameAddress());
+ assertFalse("i2 attempt 4 must exhaust", i2.stillTheSameAddress());
+ }
+
+ // -----------------------------------------------------------------------
+ // Concurrency: concurrent newStrategy() calls on a shared template
+ // -----------------------------------------------------------------------
+
+ @Test
+ public void testConcurrentPerInvocationInstancesAreIndependent() throws
InterruptedException {
+ final int maxRetries = 4;
+ RetryStrategy template = strategyWith(maxRetries);
+ AtomicInteger successes1 = new AtomicInteger();
+ AtomicInteger successes2 = new AtomicInteger();
+ CountDownLatch start = new CountDownLatch(1);
+ CountDownLatch done = new CountDownLatch(2);
+
+ for (AtomicInteger counter : new AtomicInteger[]{successes1,
successes2}) {
+ ExecutorService pool = Executors.newSingleThreadExecutor();
+ pool.submit(() -> {
+ try {
+ start.await();
+ // Simulates FailoverTargetSelector calling newStrategy()
per invocation.
+ RetryStrategy instance = (RetryStrategy)
template.newStrategy();
+ for (int i = 0; i < maxRetries; i++) {
+ if (instance.stillTheSameAddress()) {
+ counter.incrementAndGet();
+ }
+ }
+ assertFalse("instance must be exhausted",
instance.stillTheSameAddress());
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } finally {
+ done.countDown();
+ }
+ });
+ pool.shutdown();
+ }
+
+ start.countDown();
+ done.await();
+ assertEquals("invocation 1 retry count", maxRetries, successes1.get());
+ assertEquals("invocation 2 retry count", maxRetries, successes2.get());
+ }
+}
diff --git
a/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/failover/AbstractFailoverTest.java
b/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/failover/AbstractFailoverTest.java
index b2045f01729..0ed8a266244 100644
---
a/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/failover/AbstractFailoverTest.java
+++
b/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/failover/AbstractFailoverTest.java
@@ -30,6 +30,7 @@ import jakarta.ws.rs.core.Response;
import org.apache.cxf.Bus;
import org.apache.cxf.bus.extension.ExtensionManagerBus;
import org.apache.cxf.clustering.FailoverFeature;
+import org.apache.cxf.clustering.FailoverStrategy;
import org.apache.cxf.clustering.FailoverTargetSelector;
import org.apache.cxf.clustering.RandomStrategy;
import org.apache.cxf.clustering.RetryStrategy;
@@ -376,12 +377,45 @@ public abstract class AbstractFailoverTest extends
AbstractBusClientServerTestBa
private static final class CustomRetryStrategy extends RetryStrategy {
private int totalCount;
private Map<String, Integer> map = new HashMap<>();
+
+ @Override
+ public FailoverStrategy newStrategy() {
+ // Return a per-invocation delegate whose retry counter is
isolated but
+ // whose tracking (totalCount, map) accumulates on this shared
instance.
+ final CustomRetryStrategy master = this;
+ RetryStrategy delegate = new RetryStrategy() {
+ @Override
+ protected <T> T getNextAlternate(List<T> alternates) {
+ master.totalCount++;
+ T next = super.getNextAlternate(alternates);
+ if (next != null) {
+ String address = (String) next;
+ Integer count = master.map.get(address);
+ if (count == null) {
+ count = master.map.isEmpty() ? 1 /* count first
call */ : 0;
+ }
+ count++;
+ master.map.put(address, count);
+ }
+ return next;
+ }
+ };
+ delegate.setMaxNumberOfRetries(getMaxNumberOfRetries());
+ List<String> addresses = getAlternateAddresses(null);
+ if (addresses != null) {
+ delegate.setAlternateAddresses(addresses);
+ }
+ delegate.setDelayBetweenRetries(getDelayBetweenRetries());
+ return delegate;
+ }
+
+ // Direct use (outside FailoverTargetSelector) keeps original
behaviour.
@Override
protected <T> T getNextAlternate(List<T> alternates) {
totalCount++;
T next = super.getNextAlternate(alternates);
if (next != null) {
- String address = (String)next;
+ String address = (String) next;
Integer count = map.get(address);
if (count == null) {
count = map.isEmpty() ? 1 /* count first call */ : 0;
