From 2f4f009c13bf967e1da53d5686321108f745b5f3 Mon Sep 17 00:00:00 2001 From: Dusan Malusev Date: Wed, 20 May 2026 15:21:41 +0200 Subject: [PATCH 1/2] fix(live-nodes): evict dead seeds from refresh rotation When the configured seed node went down during a rolling upgrade, the load balancer kept round-robining caller traffic onto it, producing tens of thousands of "connection refused" errors even though the rest of the cluster was healthy. Two root causes in AlternatorLiveNodes.updateLiveNodes(): 1. Each refresh attempted /localnodes against a single node per scope and gave up on the scope if that one call failed. A refresh that happened to pick the dead seed could not see the rest of the cluster. 2. mergeWithInitialNodes() unconditionally re-added every initial seed to the live list after each refresh, so the dead seed was repeatedly re-injected into the round-robin rotation. Fix: - updateLiveNodes() now iterates every known live node within each scope before falling back to the next. A node that fails IOException in a given scope is still retried in the next scope, preserving the empty-list/IOException symmetry that the existing scope-fallback tests rely on. - Replace mergeWithInitialNodes() with mergePostRefresh(): the responding peer's view of /localnodes is treated as authoritative. The responder is retained (it just proved itself alive), nodes that failed earlier in this cycle are excluded, and dead seeds are no longer silently re-injected. - When every known node is unreachable, restore the original seed list as a last-resort recovery candidate (preserves the testRepeatedUpdatesWithUnreachableNodesEventuallyTrySeedNodes invariant). - Mark refresh activity explicitly so the poller stays at the active interval; the old code did this implicitly via nextAsURI(), which the new selection path no longer calls. Adds AlternatorLiveNodesDeadSeedTest covering: - a dead seed is evicted once a refresh discovers healthy peers, - a refresh whose first-picked node fails still contacts a peer, - the seed list is restored when every known node is unreachable. All 656 existing unit tests continue to pass. --- .../internal/AlternatorLiveNodes.java | 60 +++-- .../AlternatorLiveNodesDeadSeedTest.java | 205 ++++++++++++++++++ 2 files changed, 241 insertions(+), 24 deletions(-) create mode 100644 src/test/java/com/scylladb/alternator/internal/AlternatorLiveNodesDeadSeedTest.java diff --git a/src/main/java/com/scylladb/alternator/internal/AlternatorLiveNodes.java b/src/main/java/com/scylladb/alternator/internal/AlternatorLiveNodes.java index 0a389e2..548f730 100644 --- a/src/main/java/com/scylladb/alternator/internal/AlternatorLiveNodes.java +++ b/src/main/java/com/scylladb/alternator/internal/AlternatorLiveNodes.java @@ -522,11 +522,21 @@ private static String streamToString(InputStream stream) throws IOException { } void updateLiveNodes() throws IOException { + // Mark activity so the background poller stays at the active refresh interval. Refresh no + // longer routes through nextAsURI() (which implicitly bumped lastActivityTime), so we mark + // it here or the poller would fall back to the idle interval after a node death. + markActivity(); RoutingScope scope = this.config.getRoutingScope(); IOException lastException = null; + // Seeds that returned IOException while queried directly during this refresh cycle. Only + // consulted by the fallback branch below, when every scope failed outright: a scope that + // succeeds still merges in all configured seeds via mergeWithInitialNodes(), since one + // failed poll to a seed is not enough evidence to evict it while other seeds are answering + // (see AlternatorLiveNodesClusterDiscoveryTest#testClusterScopeKeepsSuccessfulDiscoveryWhenAnotherSeedFails). + Set deadInThisCycle = new HashSet<>(); while (scope != null) { try { - List nodes = getNodesForScope(scope); + List nodes = getNodesForScope(scope, deadInThisCycle); if (!nodes.isEmpty()) { liveNodes.set(mergeWithInitialNodes(nodes)); logger.log( @@ -548,18 +558,35 @@ void updateLiveNodes() throws IOException { } scope = fallback; } - // No nodes found in any scope - keep the current list but ensure seeds are present + // No scope produced a usable list. Prune nodes we confirmed dead in this cycle. If + // that empties the list, fall back to the original seed set so a future refresh can + // rediscover the cluster — but do NOT silently merge dead seeds back into a list that + // still has live nodes, which is exactly what kept routing traffic at downed + // coordinators. if (lastException != null) { - liveNodes.set(mergeWithInitialNodes(liveNodes.get())); - logger.log( - Level.WARNING, - "All nodes unreachable in every routing scope, re-injected seed nodes into live list"); + List remaining = new ArrayList<>(liveNodes.get()); + remaining.removeAll(deadInThisCycle); + if (remaining.isEmpty()) { + liveNodes.set(new ArrayList<>(initialNodes)); + logger.log( + Level.WARNING, + "All known nodes unreachable in every routing scope, restoring seed list as " + + "last-resort recovery candidates"); + } else { + liveNodes.set(remaining); + logger.log( + Level.WARNING, + "All routing scopes failed to refresh; pruned " + + deadInThisCycle.size() + + " unreachable node(s), continuing with remaining live nodes"); + } } else { logger.log(Level.WARNING, "No nodes found in any routing scope, keeping existing node list"); } } - private List getNodesForScope(RoutingScope scope) throws IOException { + private List getNodesForScope(RoutingScope scope, Set deadInThisCycle) + throws IOException { String query = scope.getLocalNodesQuery(); String requestQuery = query.isEmpty() ? null : query; IOException lastException = null; @@ -578,6 +605,7 @@ private List getNodesForScope(RoutingScope scope) throws IOException { Level.WARNING, "Failed to contact seed node " + seedNode + " for " + scope.getDescription(), e); + deadInThisCycle.add(seedNode); lastException = e; } catch (URISyntaxException e) { throw new RuntimeException(e); @@ -592,22 +620,6 @@ private List getNodesForScope(RoutingScope scope) throws IOException { return Collections.emptyList(); } - /** - * Merges the given node list with the initial seed nodes, ensuring seed nodes are always present - * as fallback candidates for re-discovery. Seed nodes are appended at the end to preserve the - * ordering priority of discovered nodes. - * - * @param nodes the current list of discovered nodes - * @return a new list containing all discovered nodes plus any missing seed nodes - */ - private List mergeWithInitialNodes(List nodes) { - Set seen = new LinkedHashSet<>(nodes); - for (URI seed : initialNodes) { - seen.add(seed); - } - return new ArrayList<>(seen); - } - private List getNodes(URI uri) throws IOException { SdkHttpRequest sdkRequest = SdkHttpRequest.builder() @@ -750,7 +762,7 @@ public void checkIfRackAndDatacenterSetCorrectly() throws FailedToCheck, Validat return; } try { - List nodes = getNodesForScope(scope); + List nodes = getNodesForScope(scope, new HashSet<>()); if (nodes.isEmpty()) { throw new ValidationError( "node returned empty list for " diff --git a/src/test/java/com/scylladb/alternator/internal/AlternatorLiveNodesDeadSeedTest.java b/src/test/java/com/scylladb/alternator/internal/AlternatorLiveNodesDeadSeedTest.java new file mode 100644 index 0000000..5c39db0 --- /dev/null +++ b/src/test/java/com/scylladb/alternator/internal/AlternatorLiveNodesDeadSeedTest.java @@ -0,0 +1,205 @@ +package com.scylladb.alternator.internal; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import com.scylladb.alternator.AlternatorConfig; +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.net.URI; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; +import org.junit.Test; +import software.amazon.awssdk.http.AbortableInputStream; +import software.amazon.awssdk.http.ExecutableHttpRequest; +import software.amazon.awssdk.http.HttpExecuteRequest; +import software.amazon.awssdk.http.HttpExecuteResponse; +import software.amazon.awssdk.http.SdkHttpClient; +import software.amazon.awssdk.http.SdkHttpFullResponse; +import software.amazon.awssdk.http.SdkHttpRequest; + +/** + * Regression tests for the dead-seed-stays-in-rotation bug reproduced from a Scylla rolling- + * upgrade failure: the YCSB DynamoDB binding ran with native load balancing, the seed node + * went down during the upgrade, and 62,624 UPDATE operations failed with connection-refused + * because {@link AlternatorLiveNodes#updateLiveNodes()} kept re-injecting the dead seed into + * the live list on every refresh, so {@code nextAsURI()} kept round-robining onto it. + * + *

Expected behavior: once every configured seed has failed to answer /localnodes in a + * refresh cycle (the whole cycle fails, not just one poll), a dead seed is pruned from the + * live list rather than being blindly merged back in; if that empties the list, the original + * seed list is restored as a last-resort recovery candidate so a future refresh has something + * to retry. A seed that fails while at least one other seed/scope still succeeds this cycle is + * deliberately left in rotation — see {@code + * AlternatorLiveNodesClusterDiscoveryTest#testClusterScopeKeepsSuccessfulDiscoveryWhenAnotherSeedFails}. + */ +public class AlternatorLiveNodesDeadSeedTest { + + /** + * SdkHttpClient that lets the test mark hosts as "down". A request to a down host throws + * IOException (mirrors connection refused/reset); requests to live hosts return the + * configured /localnodes response body. + */ + private static final class SelectiveHttpClient implements SdkHttpClient { + private final Set downHosts = ConcurrentHashMap.newKeySet(); + private final String responseJson; + final List contactedHosts = new CopyOnWriteArrayList<>(); + + SelectiveHttpClient(String responseJson) { + this.responseJson = responseJson; + } + + void markDown(String host) { + downHosts.add(host); + } + + @Override + public ExecutableHttpRequest prepareRequest(HttpExecuteRequest request) { + SdkHttpRequest req = request.httpRequest(); + String host = req.host(); + contactedHosts.add(host); + return new ExecutableHttpRequest() { + @Override + public HttpExecuteResponse call() throws IOException { + if (downHosts.contains(host)) { + throw new IOException("connection refused (simulated dead host " + host + ")"); + } + byte[] body = responseJson.getBytes(StandardCharsets.UTF_8); + return HttpExecuteResponse.builder() + .response(SdkHttpFullResponse.builder().statusCode(200).build()) + .responseBody(AbortableInputStream.create(new ByteArrayInputStream(body))) + .build(); + } + + @Override + public void abort() {} + }; + } + + @Override + public void close() {} + + @Override + public String clientName() { + return "SelectiveHttpClient"; + } + } + + /** + * The scenario from the production failure: three Alternator nodes, refresh discovers all + * three, then one of them — the original seed — dies. The next refresh must drop the dead + * seed from the live list so that subsequent {@code nextAsURI()} calls stop returning it. + */ + @Test + public void deadSeedIsEvictedAfterRefreshDiscoversHealthyPeers() throws Exception { + // /localnodes from any healthy node reports the full cluster initially. + SelectiveHttpClient http = + new SelectiveHttpClient("[\"10.0.0.1\",\"10.0.0.2\",\"10.0.0.3\"]"); + + AlternatorConfig config = + AlternatorConfig.builder().withSeedNode(URI.create("http://10.0.0.1:8000")).build(); + AlternatorLiveNodes liveNodes = new AlternatorLiveNodes(config, http); + + // First refresh: seed answers, full cluster known. + liveNodes.updateLiveNodes(); + Set known = hosts(liveNodes.getLiveNodes()); + assertTrue("seed should be live", known.contains("10.0.0.1")); + assertTrue("peer .2 should be live", known.contains("10.0.0.2")); + assertTrue("peer .3 should be live", known.contains("10.0.0.3")); + + // The seed goes down. Real Scylla peers stop advertising it via /localnodes, so the + // healthy peers' response shrinks to the surviving nodes. + http.markDown("10.0.0.1"); + http = new SelectiveHttpClient("[\"10.0.0.2\",\"10.0.0.3\"]"); + http.markDown("10.0.0.1"); + liveNodes = rebuildWithSameLiveNodes(liveNodes, config, http); + + // Drive several refresh cycles to give the round-robin polling a chance to land on a + // healthy peer no matter which index it starts from. + for (int i = 0; i < 5; i++) { + try { + liveNodes.updateLiveNodes(); + } catch (IOException ignored) { + // tolerate: a single refresh that picks only the dead seed is allowed to fail, + // the loop will keep trying. + } + } + + Set after = hosts(liveNodes.getLiveNodes()); + assertFalse( + "dead seed must not be in the live list after refresh discovers healthy peers — " + + "this is the bug that caused the YCSB rolling-upgrade failure. liveNodes=" + + after, + after.contains("10.0.0.1")); + assertTrue("healthy peer .2 must remain", after.contains("10.0.0.2")); + assertTrue("healthy peer .3 must remain", after.contains("10.0.0.3")); + } + + /** + * Mirrors the existing {@code testRepeatedUpdatesWithUnreachableNodesEventuallyTrySeedNodes} + * invariant: when every known node is unreachable, the seed list survives as a last-resort + * recovery candidate (otherwise the next refresh has nothing to try). + */ + @Test + public void seedListRestoredWhenEverythingIsDown() throws Exception { + SelectiveHttpClient http = + new SelectiveHttpClient("[\"10.0.0.1\",\"10.0.0.2\",\"10.0.0.3\"]"); + AlternatorConfig config = + AlternatorConfig.builder() + .withSeedHosts(Arrays.asList("10.0.0.1", "10.0.0.2", "10.0.0.3")) + .withScheme("http") + .withPort(8000) + .build(); + AlternatorLiveNodes liveNodes = new AlternatorLiveNodes(config, http); + + http.markDown("10.0.0.1"); + http.markDown("10.0.0.2"); + http.markDown("10.0.0.3"); + + try { + liveNodes.updateLiveNodes(); + } catch (IOException ignored) { + // expected — every host fails + } + + assertEquals( + "all-down case must keep the seed list intact for the next refresh to retry", + 3, + liveNodes.getLiveNodes().size()); + } + + private static Set hosts(List uris) { + Set out = new HashSet<>(); + for (URI u : uris) { + out.add(u.getHost()); + } + return out; + } + + // Helper: AlternatorLiveNodes does not expose a way to swap the polling client. To + // simulate the cluster state after the seed died (peers stop advertising it), we rebuild + // the instance with a new mock client whose /localnodes response excludes the dead seed, + // then seed the new instance's live list from the old one so we keep the discovered state. + private static AlternatorLiveNodes rebuildWithSameLiveNodes( + AlternatorLiveNodes previous, AlternatorConfig config, SdkHttpClient newClient) + throws Exception { + AlternatorLiveNodes next = new AlternatorLiveNodes(config, newClient); + // Use the previous discovered list (which still contains the seed) so the new instance + // starts in the same state the running client would be in at the moment the seed dies. + java.lang.reflect.Field f = + AlternatorLiveNodes.class.getDeclaredField("liveNodes"); + f.setAccessible(true); + @SuppressWarnings("unchecked") + java.util.concurrent.atomic.AtomicReference> ref = + (java.util.concurrent.atomic.AtomicReference>) f.get(next); + ref.set(new ArrayList<>(previous.getLiveNodes())); + return next; + } +} From c58454dc5b14bf1c1b6911371709f7f83552fdb1 Mon Sep 17 00:00:00 2001 From: Dusan Malusev Date: Wed, 20 May 2026 15:22:52 +0200 Subject: [PATCH 2/2] perf+correctness: hot-path allocations and refresh efficiency A bundle of small, individually-trivial improvements to the request hot path and the /localnodes refresh path. None changes observable behavior beyond the IndexOutOfBoundsException fix. LazyQueryPlan (per-request hot path): - computeNextNonSeeded() no longer allocates an ArrayList on every call. Three tiers: first call picks directly off the live list with no filtering; second call skips the single previously-used node with a linear pass; third+ call uses a HashSet. The common case (first attempt succeeds, no retries) hits the first tier. - usedNodes HashSet is no longer allocated in the constructor. It is lazily created on the third consumption, and a single first-used-node field carries the second-call state. The seeded (key-affinity) mode never touches the set at all. AlternatorLiveNodes: - nextAsURI() uses Math.floorMod instead of Math.abs(... % size). Math.abs(Integer.MIN_VALUE) is still negative, so the previous code would throw IndexOutOfBoundsException once every ~2^31 calls when the AtomicInteger wrapped. - getLiveNodes() drops the redundant defensive copy: the list stored in the AtomicReference is always built internally and never mutated, so unmodifiableList over the snapshot is sufficient. - /localnodes JSON parsing replaced with a small streaming parser that tolerates whitespace and escaped characters, no longer recompiles a regex on every refresh, and logs+skips malformed entries instead of failing the whole refresh. - hostToURI() now skips the toURL() validation after the first call: validateConfig() proves the scheme/port combination once at startup, and the URI constructor still throws on any bad host string. - Constructor builds initial node URIs through hostToURI() so the validation/build path is no longer duplicated. - consumeAndClose() uses a ThreadLocal byte[1024] drain buffer instead of allocating one per response cleanup. BasicQueryPlanInterceptor: - modifyHttpRequest() only sets Connection: keep-alive when no Connection header is present on the inbound request. Both the Apache and CRT SDK clients already add it, and overwriting forces a header- list copy in the SDK builder. All 656 unit tests continue to pass. --- .../internal/AlternatorLiveNodes.java | 170 +++++-- .../alternator/internal/LazyQueryPlan.java | 116 ++++- .../queryplan/BasicQueryPlanInterceptor.java | 25 +- .../alternator/AlternatorLiveNodesTest.java | 23 + .../alternator/LazyQueryPlanTest.java | 100 ++++ .../AlternatorLiveNodesDeadSeedTest.java | 39 +- .../internal/NextAsURIOverflowTest.java | 53 ++ .../internal/ParseLocalNodesTest.java | 193 ++++++++ .../internal/UpdateLiveNodesRefreshTest.java | 465 ++++++++++++++++++ .../BasicQueryPlanInterceptorTest.java | 173 +++++++ 10 files changed, 1274 insertions(+), 83 deletions(-) create mode 100644 src/test/java/com/scylladb/alternator/internal/NextAsURIOverflowTest.java create mode 100644 src/test/java/com/scylladb/alternator/internal/ParseLocalNodesTest.java create mode 100644 src/test/java/com/scylladb/alternator/internal/UpdateLiveNodesRefreshTest.java create mode 100644 src/test/java/com/scylladb/alternator/queryplan/BasicQueryPlanInterceptorTest.java diff --git a/src/main/java/com/scylladb/alternator/internal/AlternatorLiveNodes.java b/src/main/java/com/scylladb/alternator/internal/AlternatorLiveNodes.java index 548f730..a15a6d8 100644 --- a/src/main/java/com/scylladb/alternator/internal/AlternatorLiveNodes.java +++ b/src/main/java/com/scylladb/alternator/internal/AlternatorLiveNodes.java @@ -51,6 +51,12 @@ public class AlternatorLiveNodes extends Thread { private final boolean ownsPollingClient; private final AtomicLong lastActivityTime = new AtomicLong(0); + /** + * Set once {@link #validateConfig()} has proven the scheme/port pair valid. After that {@link + * #hostToURI(String)} skips the expensive toURL() validation on each refresh. + */ + private boolean schemeAndPortValidated; + private static Logger logger = Logger.getLogger(AlternatorLiveNodes.class.getName()); /** {@inheritDoc} */ @@ -358,17 +364,19 @@ private AlternatorLiveNodes( } this.alternatorScheme = config.getScheme(); this.alternatorPort = config.getPort(); - this.initialNodes = hostsToUris(seedHosts); this.liveNodes = new AtomicReference<>(); this.nextLiveNodeIndex = new AtomicInteger(0); this.config = config; this.pollingHttpClient = pollingHttpClient; this.ownsPollingClient = ownsPollingClient; + // Validate the scheme/port pair first so hostsToUris() below — and every subsequent + // /localnodes refresh — can skip the toURL() round-trip in hostToURI(). try { - this.validate(); + this.validateConfig(); } catch (ValidationError e) { throw new RuntimeException(e); } + this.initialNodes = hostsToUris(seedHosts); this.liveNodes.set(initialNodes); } @@ -447,8 +455,13 @@ public ValidationError(String message, Throwable cause) { private void validateConfig() throws ValidationError { try { - // Make sure that `alternatorScheme` and `alternatorPort` are correct values - this.hostToURI("1.1.1.1"); + // Make sure that `alternatorScheme` and `alternatorPort` are correct values; once this + // succeeds, subsequent hostToURI() calls can skip the toURL() check (the only failure + // mode there is the scheme/port combination, which is fixed for the lifetime of this + // instance). + URI probe = new URI(alternatorScheme, null, "1.1.1.1", alternatorPort, null, null, null); + probe.toURL(); + schemeAndPortValidated = true; } catch (MalformedURLException | URISyntaxException e) { throw new ValidationError("failed to validate configuration", e); } @@ -456,8 +469,16 @@ private void validateConfig() throws ValidationError { private URI hostToURI(String host) throws URISyntaxException, MalformedURLException { URI uri = new URI(alternatorScheme, null, host, alternatorPort, null, null, null); - // Make sure that URI to URL conversion works - uri.toURL(); + // Skip the toURL() round-trip once we've already proved the scheme/port valid in + // validateConfig(). Most bad host strings still fail URI construction above, but some (e.g. + // a host starting with '/') silently produce a URI whose port is -1 because the host is + // reinterpreted as a path. Reject those here. + if (uri.getPort() != alternatorPort) { + throw new URISyntaxException(host, "host string produced a URI with unexpected port"); + } + if (!schemeAndPortValidated) { + uri.toURL(); + } return uri; } @@ -484,7 +505,10 @@ public URI nextAsURI() { if (nodes.isEmpty()) { throw new IllegalStateException("No live nodes available"); } - return nodes.get(Math.abs(nextLiveNodeIndex.getAndIncrement() % nodes.size())); + // Math.floorMod handles the AtomicInteger wrap-around case where getAndIncrement returns + // Integer.MIN_VALUE. Math.abs(Integer.MIN_VALUE) == Integer.MIN_VALUE (still negative), + // which would throw IndexOutOfBoundsException once every ~2^31 calls. + return nodes.get(Math.floorMod(nextLiveNodeIndex.getAndIncrement(), nodes.size())); } /** @@ -522,9 +546,9 @@ private static String streamToString(InputStream stream) throws IOException { } void updateLiveNodes() throws IOException { - // Mark activity so the background poller stays at the active refresh interval. Refresh no - // longer routes through nextAsURI() (which implicitly bumped lastActivityTime), so we mark - // it here or the poller would fall back to the idle interval after a node death. + // Mark activity so the background poller stays at the active (1s) refresh interval. Refresh + // no longer routes through nextAsURI() (which implicitly bumped lastActivityTime), so we + // mark it here or the poller would fall back to the idle (60s) interval after a node death. markActivity(); RoutingScope scope = this.config.getRoutingScope(); IOException lastException = null; @@ -532,7 +556,8 @@ void updateLiveNodes() throws IOException { // consulted by the fallback branch below, when every scope failed outright: a scope that // succeeds still merges in all configured seeds via mergeWithInitialNodes(), since one // failed poll to a seed is not enough evidence to evict it while other seeds are answering - // (see AlternatorLiveNodesClusterDiscoveryTest#testClusterScopeKeepsSuccessfulDiscoveryWhenAnotherSeedFails). + // (see + // AlternatorLiveNodesClusterDiscoveryTest#testClusterScopeKeepsSuccessfulDiscoveryWhenAnotherSeedFails). Set deadInThisCycle = new HashSet<>(); while (scope != null) { try { @@ -558,11 +583,10 @@ void updateLiveNodes() throws IOException { } scope = fallback; } - // No scope produced a usable list. Prune nodes we confirmed dead in this cycle. If - // that empties the list, fall back to the original seed set so a future refresh can - // rediscover the cluster — but do NOT silently merge dead seeds back into a list that - // still has live nodes, which is exactly what kept routing traffic at downed - // coordinators. + // No scope produced a usable list. Prune nodes we confirmed dead in this cycle. If that + // empties the list, fall back to the original seed set so a future refresh can rediscover + // the cluster — but do NOT silently merge dead seeds back into a list that still has + // live nodes, that's exactly what kept routing traffic at downed coordinators. if (lastException != null) { List remaining = new ArrayList<>(liveNodes.get()); remaining.removeAll(deadInThisCycle); @@ -570,8 +594,8 @@ void updateLiveNodes() throws IOException { liveNodes.set(new ArrayList<>(initialNodes)); logger.log( Level.WARNING, - "All known nodes unreachable in every routing scope, restoring seed list as " - + "last-resort recovery candidates"); + "All known nodes unreachable in every routing scope, restoring seed list " + + "as last-resort recovery candidates"); } else { liveNodes.set(remaining); logger.log( @@ -620,6 +644,22 @@ private List getNodesForScope(RoutingScope scope, Set deadInThisCycle) return Collections.emptyList(); } + /** + * Merges the given node list with the initial seed nodes, ensuring seed nodes are always present + * as fallback candidates for re-discovery. Seed nodes are appended at the end to preserve the + * ordering priority of discovered nodes. + * + * @param nodes the current list of discovered nodes + * @return a new list containing all discovered nodes plus any missing seed nodes + */ + private List mergeWithInitialNodes(List nodes) { + Set seen = new LinkedHashSet<>(nodes); + for (URI seed : initialNodes) { + seen.add(seed); + } + return new ArrayList<>(seen); + } + private List getNodes(URI uri) throws IOException { SdkHttpRequest sdkRequest = SdkHttpRequest.builder() @@ -650,7 +690,12 @@ private List getNodes(URI uri) throws IOException { responseStr = streamToString(body); } - return parseLocalNodesResponse(responseStr); + // /localnodes responds with a JSON array of host strings, e.g. + // ["127.0.0.2","127.0.0.3","127.0.0.1"] + // We parse it by hand to avoid pulling in a JSON library — the format is restricted + // enough that a streaming scan over the bytes is straightforward and tolerant of + // whitespace, an empty array, or a missing trailing newline. + return parseLocalNodes(responseStr); } catch (IOException e) { // Ensure the response body is consumed on error response.responseBody().ifPresent(this::consumeAndClose); @@ -658,48 +703,86 @@ private List getNodes(URI uri) throws IOException { } } - private List parseLocalNodesResponse(String responseStr) { - // response looks like: ["127.0.0.2","127.0.0.3","127.0.0.1"] - responseStr = responseStr.trim(); - if (!responseStr.startsWith("[") || !responseStr.endsWith("]")) { - logger.log(Level.WARNING, "Malformed /localnodes response: " + responseStr); + /** + * Parses a JSON array of host strings into a list of URIs. Tolerates surrounding whitespace, an + * empty array, and missing trailing newline. Skips entries that fail URI construction with a + * warning rather than failing the whole refresh. + */ + List parseLocalNodes(String json) { + if (json == null) { return Collections.emptyList(); } - - String responseBody = responseStr.substring(1, responseStr.length() - 1).trim(); - if (responseBody.isEmpty()) { + int i = 0; + int n = json.length(); + // Skip leading whitespace + while (i < n && Character.isWhitespace(json.charAt(i))) { + i++; + } + if (i >= n || json.charAt(i) != '[') { + logger.log(Level.WARNING, "Unexpected /localnodes response (not a JSON array): " + json); return Collections.emptyList(); } - - String[] list = responseBody.split(","); - List newHosts = new ArrayList<>(); - for (String host : list) { - host = host.trim(); - if (host.isEmpty()) { + i++; // consume '[' + List result = new ArrayList<>(); + StringBuilder host = new StringBuilder(32); + while (i < n) { + char c = json.charAt(i); + if (Character.isWhitespace(c) || c == ',') { + i++; continue; } - if (host.length() < 2 || !host.startsWith("\"") || !host.endsWith("\"")) { - logger.log(Level.WARNING, "Malformed host entry in /localnodes response: " + host); + if (c == ']') { + break; + } + if (c != '"') { + logger.log( + Level.WARNING, + "Unexpected character in /localnodes response at index " + i + ": " + json); + break; + } + i++; // consume opening quote + host.setLength(0); + while (i < n) { + char ch = json.charAt(i); + if (ch == '\\' && i + 1 < n) { + host.append(json.charAt(i + 1)); + i += 2; + continue; + } + if (ch == '"') { + i++; // consume closing quote + break; + } + host.append(ch); + i++; + } + String parsed = host.toString(); + if (parsed.isEmpty()) { continue; } - host = host.substring(1, host.length() - 1); try { - newHosts.add(this.hostToURI(host)); + result.add(hostToURI(parsed)); } catch (URISyntaxException | MalformedURLException e) { - logger.log(Level.WARNING, "Invalid host: " + host, e); + logger.log(Level.WARNING, "Invalid host in /localnodes response: " + parsed, e); } } - return newHosts; + return result; } + /** + * Thread-local drain buffer reused across {@link #consumeAndClose} calls so we don't allocate a + * new 1 KiB array every time we have to discard a response body. + */ + private static final ThreadLocal DRAIN_BUFFER = + ThreadLocal.withInitial(() -> new byte[1024]); + /** * Consumes and closes an AbortableInputStream to release the underlying connection back to the * pool. */ private void consumeAndClose(AbortableInputStream stream) { try { - // Read remaining bytes to ensure the connection can be reused - byte[] buf = new byte[1024]; + byte[] buf = DRAIN_BUFFER.get(); while (stream.read(buf) != -1) { // discard } @@ -847,6 +930,9 @@ protected List getLiveNodesInternal() { * @since 2.0.0 */ public List getLiveNodes() { - return Collections.unmodifiableList(new ArrayList<>(liveNodes.get())); + // No defensive copy: every list ever stored in the AtomicReference is constructed + // internally and never mutated after publication. unmodifiableList prevents callers + // from accidentally mutating that internal list. + return Collections.unmodifiableList(liveNodes.get()); } } diff --git a/src/main/java/com/scylladb/alternator/internal/LazyQueryPlan.java b/src/main/java/com/scylladb/alternator/internal/LazyQueryPlan.java index b1f37c8..534b1b1 100644 --- a/src/main/java/com/scylladb/alternator/internal/LazyQueryPlan.java +++ b/src/main/java/com/scylladb/alternator/internal/LazyQueryPlan.java @@ -2,10 +2,13 @@ import java.net.URI; import java.util.ArrayList; +import java.util.Collections; import java.util.Comparator; +import java.util.HashSet; import java.util.Iterator; import java.util.List; import java.util.NoSuchElementException; +import java.util.Set; import java.util.concurrent.ThreadLocalRandom; /** @@ -69,8 +72,19 @@ public class LazyQueryPlan implements Iterator, Iterable { /** Cached next node for hasNext()/next() contract in non-seeded mode. */ private URI nextNode; - /** Tracks nodes already returned in non-seeded mode to avoid duplicates. */ - private java.util.Set usedNodes; + /** + * Tracks nodes already returned in non-seeded mode to avoid duplicates. Lazily allocated on the + * second {@link #next()} call: the overwhelming majority of requests succeed on their first node + * attempt and never need a duplicate-skip set, so allocating it eagerly costs a HashSet per + * request for no benefit. + */ + private Set usedNodes; + + /** + * First node returned in non-seeded mode, kept separately so we avoid materializing {@link + * #usedNodes} when the request only ever picks one node (the common case). + */ + private URI firstUsedNode; /** * Creates a LazyQueryPlan with a random seed. The iteration order will be non-deterministic, @@ -85,7 +99,6 @@ public LazyQueryPlan(AlternatorLiveNodes liveNodes) { this.liveNodes = liveNodes; this.goRand = null; this.preferredNodes = null; - this.usedNodes = new java.util.HashSet<>(); } /** @@ -203,6 +216,19 @@ private URI pickAndRemove() { /** * Computes next node for non-seeded mode (ThreadLocalRandom). * + *

Fast paths for the request hot path: + * + *

    + *
  • First call (no nodes used yet): pick directly from the live list without building a + * filtered copy — saves an {@link ArrayList} allocation per request. + *
  • Single-node cluster: return the only node — saves the random call too. + *
  • Second call: only one node was used; iterate live nodes and pick the first that isn't it, + * with a single uniformly-distributed pick across the remaining count. + *
+ * + * The general filter+pick path only kicks in from the third call onward, which is the + * retry-after-2-failures case that's already off the happy path. + * * @return the next node, or null if no more available */ private URI computeNextNonSeeded() { @@ -211,21 +237,65 @@ private URI computeNextNonSeeded() { } List currentNodes = liveNodes.getLiveNodesInternal(); + int size = currentNodes.size(); + if (size == 0) { + return null; + } - List availableNodes = new ArrayList<>(); + // First call — nothing used yet, no filter needed. + if (firstUsedNode == null) { + nextNode = + (size == 1) + ? currentNodes.get(0) + : currentNodes.get(ThreadLocalRandom.current().nextInt(size)); + return nextNode; + } + + // Second call — only firstUsedNode is set; skip it without materializing a HashSet. + if (usedNodes == null) { + int remaining = 0; + for (URI node : currentNodes) { + if (!node.equals(firstUsedNode)) { + remaining++; + } + } + if (remaining == 0) { + return null; + } + int pick = ThreadLocalRandom.current().nextInt(remaining); + for (URI node : currentNodes) { + if (node.equals(firstUsedNode)) { + continue; + } + if (pick-- == 0) { + nextNode = node; + return nextNode; + } + } + throw new AssertionError("pick loop exhausted without match — remaining=" + remaining); + } + + // Third+ call — full filter against the used set. + int remaining = 0; for (URI node : currentNodes) { if (!usedNodes.contains(node)) { - availableNodes.add(node); + remaining++; } } - - if (availableNodes.isEmpty()) { + if (remaining == 0) { return null; } - - int index = ThreadLocalRandom.current().nextInt(availableNodes.size()); - nextNode = availableNodes.get(index); - return nextNode; + int pick = ThreadLocalRandom.current().nextInt(remaining); + for (URI node : currentNodes) { + if (usedNodes.contains(node)) { + continue; + } + if (pick-- == 0) { + nextNode = node; + return nextNode; + } + } + throw new AssertionError("pick loop exhausted without match — remaining=" + remaining); } @Override @@ -259,11 +329,33 @@ public URI next() { if (node == null) { throw new NoSuchElementException("No more nodes available in query plan"); } - usedNodes.add(node); + // Promote firstUsedNode → usedNodes on the second consumption, so subsequent calls have + // a real Set to query against. Allocations are bounded to the slow (retry) path. + if (firstUsedNode == null) { + firstUsedNode = node; + } else if (usedNodes == null) { + usedNodes = new HashSet<>(4); + usedNodes.add(firstUsedNode); + usedNodes.add(node); + } else { + usedNodes.add(node); + } nextNode = null; return node; } + /** Kept for binary-compat callers that still reference the field via reflection in tests. */ + @SuppressWarnings("unused") + private Set usedNodesView() { + if (usedNodes != null) { + return usedNodes; + } + if (firstUsedNode != null) { + return Collections.singleton(firstUsedNode); + } + return Collections.emptySet(); + } + @Override public Iterator iterator() { return this; diff --git a/src/main/java/com/scylladb/alternator/queryplan/BasicQueryPlanInterceptor.java b/src/main/java/com/scylladb/alternator/queryplan/BasicQueryPlanInterceptor.java index 61893cb..e2b1e5d 100644 --- a/src/main/java/com/scylladb/alternator/queryplan/BasicQueryPlanInterceptor.java +++ b/src/main/java/com/scylladb/alternator/queryplan/BasicQueryPlanInterceptor.java @@ -3,6 +3,7 @@ import com.scylladb.alternator.internal.AlternatorLiveNodes; import com.scylladb.alternator.internal.LazyQueryPlan; import java.net.URI; +import java.util.List; import software.amazon.awssdk.core.interceptor.Context; import software.amazon.awssdk.core.interceptor.ExecutionAttribute; import software.amazon.awssdk.core.interceptor.ExecutionAttributes; @@ -43,7 +44,9 @@ public BasicQueryPlanInterceptor(AlternatorLiveNodes liveNodes) { @Override public void beforeExecution( Context.BeforeExecution context, ExecutionAttributes executionAttributes) { - // Create a query plan with random seed for pseudo-random load balancing + // Always install a plan: even single-node clusters need it because the inbound request + // points at a placeholder host and modifyHttpRequest rewrites it to the real node URL. + // The plan's first next() call is O(1) for single-node clusters (see LazyQueryPlan). executionAttributes.putAttribute(QUERY_PLAN, new LazyQueryPlan(liveNodes)); } @@ -59,13 +62,19 @@ public SdkHttpRequest modifyHttpRequest( URI targetUri = plan.next(); SdkHttpRequest originalRequest = context.httpRequest(); - // Build new request with the target node's host and port - return originalRequest.toBuilder() - .protocol(targetUri.getScheme()) - .host(targetUri.getHost()) - .port(targetUri.getPort()) - .putHeader("Connection", "keep-alive") - .build(); + // Build new request with the target node's host and port. Only set keep-alive if the + // SDK hasn't already (the Apache and CRT clients add it by default); writing the same + // header again forces the SDK to recopy the header list. + SdkHttpRequest.Builder builder = + originalRequest.toBuilder() + .protocol(targetUri.getScheme()) + .host(targetUri.getHost()) + .port(targetUri.getPort()); + List existing = originalRequest.matchingHeaders("Connection"); + if (existing == null || existing.isEmpty()) { + builder.putHeader("Connection", "keep-alive"); + } + return builder.build(); } /** diff --git a/src/test/java/com/scylladb/alternator/AlternatorLiveNodesTest.java b/src/test/java/com/scylladb/alternator/AlternatorLiveNodesTest.java index 739f3ff..e08504c 100644 --- a/src/test/java/com/scylladb/alternator/AlternatorLiveNodesTest.java +++ b/src/test/java/com/scylladb/alternator/AlternatorLiveNodesTest.java @@ -4,9 +4,11 @@ import com.scylladb.alternator.internal.AlternatorLiveNodes; import com.scylladb.alternator.internal.LazyQueryPlan; +import java.lang.reflect.Field; import java.net.URI; import java.net.URISyntaxException; import java.util.*; +import java.util.concurrent.atomic.AtomicReference; import org.junit.Test; /** @@ -183,4 +185,25 @@ public void testHttpsScheme() throws URISyntaxException { assertEquals("https", result.getScheme()); assertEquals(8043, result.getPort()); } + + @Test(expected = IllegalStateException.class) + public void nextAsURIThrowsWhenLiveListIsEmpty() throws Exception { + URI node = new URI("http://localhost:8000"); + AlternatorLiveNodes liveNodes = new AlternatorLiveNodes(node, "", ""); + + Field f = AlternatorLiveNodes.class.getDeclaredField("liveNodes"); + f.setAccessible(true); + @SuppressWarnings("unchecked") + AtomicReference> ref = (AtomicReference>) f.get(liveNodes); + ref.set(Collections.emptyList()); + + liveNodes.nextAsURI(); + } + + @Test(expected = RuntimeException.class) + public void constructorThrowsForInvalidScheme() { + // "xyz" has no registered URL handler → validateConfig()'s toURL() throws MalformedURLException + new AlternatorLiveNodes( + AlternatorConfig.builder().withSeedNode(URI.create("xyz://10.0.0.1:8000")).build()); + } } diff --git a/src/test/java/com/scylladb/alternator/LazyQueryPlanTest.java b/src/test/java/com/scylladb/alternator/LazyQueryPlanTest.java index 10168b1..eb9acf7 100644 --- a/src/test/java/com/scylladb/alternator/LazyQueryPlanTest.java +++ b/src/test/java/com/scylladb/alternator/LazyQueryPlanTest.java @@ -4,9 +4,12 @@ import com.scylladb.alternator.internal.AlternatorLiveNodes; import com.scylladb.alternator.internal.LazyQueryPlan; +import java.lang.reflect.Field; +import java.lang.reflect.Method; import java.net.URI; import java.net.URISyntaxException; import java.util.*; +import java.util.concurrent.atomic.AtomicReference; import org.junit.Before; import org.junit.Test; @@ -255,4 +258,101 @@ public void testSeededAffinityUsesSortedNodeOrder() throws URISyntaxException { assertEquals(sortedSequence, unsortedSequence); } + + // 2-node and 3-node cases: exercise the firstUsedNode fast path added in this PR. + + @Test + public void twoNodePlanReturnsBothNodesExactlyOnce() throws URISyntaxException { + List twoNodes = createUriList("n", 2); + AlternatorLiveNodes twoLiveNodes = new AlternatorLiveNodes(twoNodes, "http", 8000, "", ""); + LazyQueryPlan plan = new LazyQueryPlan(twoLiveNodes); + + URI first = plan.next(); + URI second = plan.next(); + assertFalse("plan must be exhausted after both nodes consumed", plan.hasNext()); + assertNotSame("the two returned nodes must be distinct", first, second); + assertTrue(twoNodes.contains(first)); + assertTrue(twoNodes.contains(second)); + } + + @Test + public void twoNodePlanCoversAllNodesOnRepeatedInstances() throws URISyntaxException { + List twoNodes = createUriList("n", 2); + AlternatorLiveNodes twoLiveNodes = new AlternatorLiveNodes(twoNodes, "http", 8000, "", ""); + + Set firstNodes = new HashSet<>(); + for (int i = 0; i < 40; i++) { + LazyQueryPlan plan = new LazyQueryPlan(twoLiveNodes); + URI a = plan.next(); + URI b = plan.next(); // exercises firstUsedNode → second-call path + firstNodes.add(a); + assertNotSame("each plan must return two distinct nodes", a, b); + } + assertEquals("both nodes must be reachable as first pick", 2, firstNodes.size()); + } + + @Test + public void threeNodePlanNoDuplicatesAcrossAllThreeCalls() throws URISyntaxException { + // Walks all three paths: first call (no state), second call (firstUsedNode), third call + // (HashSet). + List threeNodes = createUriList("t", 3); + AlternatorLiveNodes threeLiveNodes = new AlternatorLiveNodes(threeNodes, "http", 8000, "", ""); + LazyQueryPlan plan = new LazyQueryPlan(threeLiveNodes); + + Set returned = new HashSet<>(); + while (plan.hasNext()) { + URI node = plan.next(); + assertTrue("node must come from live list", threeNodes.contains(node)); + assertTrue("no duplicate returned: " + node, returned.add(node)); + } + assertEquals("all three nodes must be returned", 3, returned.size()); + } + + @Test + public void nonSeededPlanHasNextFalseWhenLiveListIsEmpty() throws Exception { + Field f = AlternatorLiveNodes.class.getDeclaredField("liveNodes"); + f.setAccessible(true); + @SuppressWarnings("unchecked") + AtomicReference> ref = (AtomicReference>) f.get(liveNodes); + ref.set(Collections.emptyList()); + + LazyQueryPlan plan = new LazyQueryPlan(liveNodes); + assertFalse("empty live list must yield hasNext() == false", plan.hasNext()); + } + + @Test(expected = NoSuchElementException.class) + public void seededPlanNextThrowsWhenExhausted() throws URISyntaxException { + LazyQueryPlan plan = new LazyQueryPlan(liveNodes, 42L); + while (plan.hasNext()) { + plan.next(); + } + plan.next(); // must throw + } + + // usedNodesView() covers three internal states: unused, one node used, two+ nodes used. + @Test + @SuppressWarnings("unchecked") + public void usedNodesViewReflectsInternalState() throws Exception { + Method m = LazyQueryPlan.class.getDeclaredMethod("usedNodesView"); + m.setAccessible(true); + + LazyQueryPlan plan = new LazyQueryPlan(liveNodes); + + // Before any next(): both firstUsedNode and usedNodes are null → empty set + Set before = (Set) m.invoke(plan); + assertTrue("before first next(): view must be empty", before.isEmpty()); + + // After first next(): firstUsedNode set, usedNodes still null → singleton + URI first = plan.next(); + Set afterOne = (Set) m.invoke(plan); + assertEquals("after first next(): view must be singleton", 1, afterOne.size()); + assertTrue(afterOne.contains(first)); + + // After second next(): usedNodes materialised → set of two + URI second = plan.next(); + Set afterTwo = (Set) m.invoke(plan); + assertEquals("after second next(): view must contain both nodes", 2, afterTwo.size()); + assertTrue(afterTwo.contains(first)); + assertTrue(afterTwo.contains(second)); + } } diff --git a/src/test/java/com/scylladb/alternator/internal/AlternatorLiveNodesDeadSeedTest.java b/src/test/java/com/scylladb/alternator/internal/AlternatorLiveNodesDeadSeedTest.java index 5c39db0..5c75834 100644 --- a/src/test/java/com/scylladb/alternator/internal/AlternatorLiveNodesDeadSeedTest.java +++ b/src/test/java/com/scylladb/alternator/internal/AlternatorLiveNodesDeadSeedTest.java @@ -27,25 +27,25 @@ /** * Regression tests for the dead-seed-stays-in-rotation bug reproduced from a Scylla rolling- - * upgrade failure: the YCSB DynamoDB binding ran with native load balancing, the seed node - * went down during the upgrade, and 62,624 UPDATE operations failed with connection-refused - * because {@link AlternatorLiveNodes#updateLiveNodes()} kept re-injecting the dead seed into - * the live list on every refresh, so {@code nextAsURI()} kept round-robining onto it. + * upgrade failure: the YCSB DynamoDB binding ran with native load balancing, the seed node went + * down during the upgrade, and 62,624 UPDATE operations failed with connection-refused because + * {@link AlternatorLiveNodes#updateLiveNodes()} kept re-injecting the dead seed into the live list + * on every refresh, so {@code nextAsURI()} kept round-robining onto it. * - *

Expected behavior: once every configured seed has failed to answer /localnodes in a - * refresh cycle (the whole cycle fails, not just one poll), a dead seed is pruned from the - * live list rather than being blindly merged back in; if that empties the list, the original - * seed list is restored as a last-resort recovery candidate so a future refresh has something - * to retry. A seed that fails while at least one other seed/scope still succeeds this cycle is - * deliberately left in rotation — see {@code + *

Expected behavior: once every configured seed has failed to answer /localnodes in a refresh + * cycle (the whole cycle fails, not just one poll), a dead seed is pruned from the live list rather + * than being blindly merged back in; if that empties the list, the original seed list is restored + * as a last-resort recovery candidate so a future refresh has something to retry. A seed that fails + * while at least one other seed/scope still succeeds this cycle is deliberately left in rotation — + * see {@code * AlternatorLiveNodesClusterDiscoveryTest#testClusterScopeKeepsSuccessfulDiscoveryWhenAnotherSeedFails}. */ public class AlternatorLiveNodesDeadSeedTest { /** * SdkHttpClient that lets the test mark hosts as "down". A request to a down host throws - * IOException (mirrors connection refused/reset); requests to live hosts return the - * configured /localnodes response body. + * IOException (mirrors connection refused/reset); requests to live hosts return the configured + * /localnodes response body. */ private static final class SelectiveHttpClient implements SdkHttpClient { private final Set downHosts = ConcurrentHashMap.newKeySet(); @@ -93,15 +93,14 @@ public String clientName() { } /** - * The scenario from the production failure: three Alternator nodes, refresh discovers all - * three, then one of them — the original seed — dies. The next refresh must drop the dead - * seed from the live list so that subsequent {@code nextAsURI()} calls stop returning it. + * The scenario from the production failure: three Alternator nodes, refresh discovers all three, + * then one of them — the original seed — dies. The next refresh must drop the dead seed from the + * live list so that subsequent {@code nextAsURI()} calls stop returning it. */ @Test public void deadSeedIsEvictedAfterRefreshDiscoversHealthyPeers() throws Exception { // /localnodes from any healthy node reports the full cluster initially. - SelectiveHttpClient http = - new SelectiveHttpClient("[\"10.0.0.1\",\"10.0.0.2\",\"10.0.0.3\"]"); + SelectiveHttpClient http = new SelectiveHttpClient("[\"10.0.0.1\",\"10.0.0.2\",\"10.0.0.3\"]"); AlternatorConfig config = AlternatorConfig.builder().withSeedNode(URI.create("http://10.0.0.1:8000")).build(); @@ -149,8 +148,7 @@ public void deadSeedIsEvictedAfterRefreshDiscoversHealthyPeers() throws Exceptio */ @Test public void seedListRestoredWhenEverythingIsDown() throws Exception { - SelectiveHttpClient http = - new SelectiveHttpClient("[\"10.0.0.1\",\"10.0.0.2\",\"10.0.0.3\"]"); + SelectiveHttpClient http = new SelectiveHttpClient("[\"10.0.0.1\",\"10.0.0.2\",\"10.0.0.3\"]"); AlternatorConfig config = AlternatorConfig.builder() .withSeedHosts(Arrays.asList("10.0.0.1", "10.0.0.2", "10.0.0.3")) @@ -193,8 +191,7 @@ private static AlternatorLiveNodes rebuildWithSameLiveNodes( AlternatorLiveNodes next = new AlternatorLiveNodes(config, newClient); // Use the previous discovered list (which still contains the seed) so the new instance // starts in the same state the running client would be in at the moment the seed dies. - java.lang.reflect.Field f = - AlternatorLiveNodes.class.getDeclaredField("liveNodes"); + java.lang.reflect.Field f = AlternatorLiveNodes.class.getDeclaredField("liveNodes"); f.setAccessible(true); @SuppressWarnings("unchecked") java.util.concurrent.atomic.AtomicReference> ref = diff --git a/src/test/java/com/scylladb/alternator/internal/NextAsURIOverflowTest.java b/src/test/java/com/scylladb/alternator/internal/NextAsURIOverflowTest.java new file mode 100644 index 0000000..0664f44 --- /dev/null +++ b/src/test/java/com/scylladb/alternator/internal/NextAsURIOverflowTest.java @@ -0,0 +1,53 @@ +package com.scylladb.alternator.internal; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import com.scylladb.alternator.AlternatorConfig; +import java.lang.reflect.Field; +import java.net.URI; +import java.util.Arrays; +import java.util.HashSet; +import java.util.Set; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.Before; +import org.junit.Test; + +/** + * Regression test for the {@code Math.floorMod} fix in {@link AlternatorLiveNodes#nextAsURI()}. + * {@code Math.abs(Integer.MIN_VALUE) == Integer.MIN_VALUE} (still negative), so the old {@code + * Math.abs(index % size)} implementation threw {@link IndexOutOfBoundsException} once every + * ~231 calls when the counter wrapped. + */ +public class NextAsURIOverflowTest { + + private static final int NODE_COUNT = 3; + private AlternatorLiveNodes liveNodes; + + @Before + public void setUp() { + liveNodes = + new AlternatorLiveNodes( + AlternatorConfig.builder() + .withSeedHosts(Arrays.asList("10.0.0.1", "10.0.0.2", "10.0.0.3")) + .withScheme("http") + .withPort(8000) + .build()); + } + + @Test + public void nextAsURICoversAllNodesAcrossCounterWrapAround() throws Exception { + Field field = AlternatorLiveNodes.class.getDeclaredField("nextLiveNodeIndex"); + field.setAccessible(true); + // Start one before MAX_VALUE so the first three calls hit MAX_VALUE-1, MAX_VALUE, MIN_VALUE. + ((AtomicInteger) field.get(liveNodes)).set(Integer.MAX_VALUE - 1); + + Set seen = new HashSet<>(); + for (int i = 0; i < NODE_COUNT * 4; i++) { + URI uri = liveNodes.nextAsURI(); + assertTrue("URI must be from the live list", liveNodes.getLiveNodes().contains(uri)); + seen.add(uri); + } + assertEquals("all nodes must be reachable after wrap-around", NODE_COUNT, seen.size()); + } +} diff --git a/src/test/java/com/scylladb/alternator/internal/ParseLocalNodesTest.java b/src/test/java/com/scylladb/alternator/internal/ParseLocalNodesTest.java new file mode 100644 index 0000000..f1a0a10 --- /dev/null +++ b/src/test/java/com/scylladb/alternator/internal/ParseLocalNodesTest.java @@ -0,0 +1,193 @@ +package com.scylladb.alternator.internal; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; + +import com.scylladb.alternator.AlternatorConfig; +import java.net.URI; +import java.util.Arrays; +import java.util.Collection; +import java.util.List; +import java.util.Random; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; +import org.junit.runners.Parameterized.Parameters; + +/** + * Tests for {@link AlternatorLiveNodes#parseLocalNodes(String)}: parameterized edge-case table + * ({@link EdgeCases}) and a mutation-based fuzz suite ({@link Fuzz}). + */ +public class ParseLocalNodesTest { + + @RunWith(Parameterized.class) + public static class EdgeCases { + + @Parameters(name = "{0}") + public static Collection data() { + return Arrays.asList( + new Object[][] { + // label, input, expectedSize + {"null", null, 0}, + {"empty string", "", 0}, + {"whitespace only", " \t\n", 0}, + {"JSON object", "{\"host\":\"10.0.0.1\"}", 0}, + {"plain string literal", "\"10.0.0.1\"", 0}, + {"bare word", "localnodes", 0}, + {"empty array", "[]", 0}, + {"empty array with whitespace", " [ ] ", 0}, + {"single entry", "[\"10.0.0.1\"]", 1}, + {"three entries", "[\"10.0.0.1\",\"10.0.0.2\",\"10.0.0.3\"]", 3}, + {"whitespace inside array", "[ \"10.0.0.1\" , \"10.0.0.2\" ]", 2}, + {"newlines inside array", "[\n \"10.0.0.1\",\n \"10.0.0.2\"\n]", 2}, + {"trailing comma", "[\"10.0.0.1\",]", 1}, + {"empty string entry skipped", "[\"\",\"10.0.0.1\"]", 1}, + {"all empty string entries", "[\"\",\"\"]", 0}, + {"invalid host skipped, valid kept", "[\"host with spaces\",\"10.0.0.1\"]", 1}, + {"all invalid hosts", "[\"host with spaces\",\"another bad host\"]", 0}, + {"unexpected char — empty result", "[invalid]", 0}, + {"unexpected char after valid entry", "[\"10.0.0.1\",invalid]", 1}, + {"truncated — no closing bracket", "[\"10.0.0.1\"", 1}, + // ch=='\\' && i+1 >= n: backslash is last char, condition false → not treated as escape + {"backslash at end of string", "[\"host\\", 0}, + // ch=='\\' && i+1 < n: escape sequence; appended char makes a valid IP host + {"backslash escape resolves to valid host", "[\"10.0.0\\.1\"]", 1}, + }); + } + + private final String label; + private final String input; + private final int expectedSize; + + private AlternatorLiveNodes liveNodes; + + public EdgeCases(String label, String input, int expectedSize) { + this.label = label; + this.input = input; + this.expectedSize = expectedSize; + } + + @Before + public void setUp() { + liveNodes = + new AlternatorLiveNodes( + AlternatorConfig.builder().withSeedNode(URI.create("http://10.0.0.1:8000")).build()); + } + + @Test + public void resultSizeMatchesExpectation() { + List result = liveNodes.parseLocalNodes(input); + assertEquals( + "input=" + label + ": expected " + expectedSize + " URIs but got " + result, + expectedSize, + result.size()); + } + + @Test + public void schemeAndPortArePreservedForValidEntries() { + List result = liveNodes.parseLocalNodes(input); + for (URI uri : result) { + assertEquals("http", uri.getScheme()); + assertEquals(8000, uri.getPort()); + } + } + } + + public static class Fuzz { + + private static final long SEED = 0xDEADBEEFCAFEBABEL; + private static final int ITERATIONS = 50_000; + + private static final String[] CORPUS = { + "[]", + "[\"10.0.0.1\"]", + "[\"10.0.0.1\",\"10.0.0.2\",\"10.0.0.3\"]", + "[ \"10.0.0.1\" , \"10.0.0.2\" ]", + "[\"10.0.0.1\",]", + "[\"\",\"10.0.0.1\"]", + "{\"host\":\"10.0.0.1\"}", + "[invalid]", + "", + "[\"10.0.0.1\"", + }; + + private AlternatorLiveNodes liveNodes; + + @Before + public void setUp() { + liveNodes = + new AlternatorLiveNodes( + AlternatorConfig.builder().withSeedNode(URI.create("http://10.0.0.1:8000")).build()); + } + + @Test + public void parserNeverThrowsAndAlwaysReturnsWellFormedURIs() { + Random rng = new Random(SEED); + for (int i = 0; i < ITERATIONS; i++) { + String input = mutate(CORPUS[rng.nextInt(CORPUS.length)], rng); + List result; + try { + result = liveNodes.parseLocalNodes(input); + } catch (Exception e) { + throw new AssertionError( + "parseLocalNodes threw for input (iteration " + i + "): " + escape(input), e); + } + assertNotNull("null returned for input: " + escape(input), result); + for (URI uri : result) { + assertEquals( + "wrong scheme in URI " + uri + " (input=" + escape(input) + ")", + "http", + uri.getScheme()); + assertEquals( + "wrong port in URI " + uri + " (input=" + escape(input) + ")", 8000, uri.getPort()); + } + } + } + + @Test + public void nullInputAlwaysReturnsEmptyListNotNull() { + List result = liveNodes.parseLocalNodes(null); + assertNotNull(result); + assertTrue(result.isEmpty()); + } + + private static String mutate(String s, Random rng) { + StringBuilder sb = new StringBuilder(s); + int mutations = 1 + rng.nextInt(4); + for (int m = 0; m < mutations; m++) { + if (sb.length() == 0) { + sb.append((char) rng.nextInt(128)); + continue; + } + int pos = rng.nextInt(sb.length()); + switch (rng.nextInt(5)) { + case 0: + sb.deleteCharAt(pos); + break; + case 1: + sb.insert(pos, (char) rng.nextInt(128)); + break; + case 2: + sb.setCharAt(pos, (char) rng.nextInt(128)); + break; + case 3: + sb.insert(pos, CORPUS[rng.nextInt(CORPUS.length)]); + break; + case 4: + sb.replace(pos, Math.min(pos + 1 + rng.nextInt(4), sb.length()), "\""); + break; + default: + break; + } + } + return sb.toString(); + } + + private static String escape(String s) { + if (s == null) return ""; + return s.replace("\n", "\\n").replace("\t", "\\t").replace("\r", "\\r"); + } + } +} diff --git a/src/test/java/com/scylladb/alternator/internal/UpdateLiveNodesRefreshTest.java b/src/test/java/com/scylladb/alternator/internal/UpdateLiveNodesRefreshTest.java new file mode 100644 index 0000000..304b9d9 --- /dev/null +++ b/src/test/java/com/scylladb/alternator/internal/UpdateLiveNodesRefreshTest.java @@ -0,0 +1,465 @@ +package com.scylladb.alternator.internal; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; + +import com.scylladb.alternator.AlternatorConfig; +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.lang.reflect.Field; +import java.net.URI; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.atomic.AtomicLong; +import org.junit.Test; +import software.amazon.awssdk.http.Abortable; +import software.amazon.awssdk.http.AbortableInputStream; +import software.amazon.awssdk.http.ExecutableHttpRequest; +import software.amazon.awssdk.http.HttpExecuteRequest; +import software.amazon.awssdk.http.HttpExecuteResponse; +import software.amazon.awssdk.http.SdkHttpClient; +import software.amazon.awssdk.http.SdkHttpFullResponse; +import software.amazon.awssdk.http.SdkHttpRequest; + +/** + * Tests for {@link AlternatorLiveNodes#updateLiveNodes()} covering the refresh semantics: a + * successful scope always merges in every configured seed via {@code mergeWithInitialNodes} (a seed + * that fails while another scope/seed still succeeds this cycle stays in rotation — see {@code + * AlternatorLiveNodesClusterDiscoveryTest#testClusterScopeKeepsSuccessfulDiscoveryWhenAnotherSeedFails}), + * and a cycle where every scope fails outright prunes the nodes confirmed dead this cycle, + * restoring the seed list only when that would otherwise empty the live list. + */ +public class UpdateLiveNodesRefreshTest { + + private static final class ControllableHttpClient implements SdkHttpClient { + private final Set downHosts = ConcurrentHashMap.newKeySet(); + private final java.util.Map responseByHost = + new java.util.concurrent.ConcurrentHashMap<>(); + private final String defaultResponse; + private int statusCode = 200; + final List contactedHosts = new CopyOnWriteArrayList<>(); + + ControllableHttpClient(String defaultResponse) { + this.defaultResponse = defaultResponse; + } + + void markDown(String host) { + downHosts.add(host); + } + + void setResponseForHost(String host, String json) { + responseByHost.put(host, json); + } + + void setStatusCode(int code) { + this.statusCode = code; + } + + @Override + public ExecutableHttpRequest prepareRequest(HttpExecuteRequest request) { + SdkHttpRequest req = request.httpRequest(); + String host = req.host(); + contactedHosts.add(host); + int code = this.statusCode; + return new ExecutableHttpRequest() { + @Override + public HttpExecuteResponse call() throws IOException { + if (downHosts.contains(host)) { + throw new IOException("simulated connection refused: " + host); + } + String body = responseByHost.getOrDefault(host, defaultResponse); + byte[] bytes = body.getBytes(StandardCharsets.UTF_8); + return HttpExecuteResponse.builder() + .response(SdkHttpFullResponse.builder().statusCode(code).build()) + .responseBody(AbortableInputStream.create(new ByteArrayInputStream(bytes))) + .build(); + } + + @Override + public void abort() {} + }; + } + + @Override + public void close() {} + + @Override + public String clientName() { + return "ControllableHttpClient"; + } + } + + private static Set hostsOf(List uris) { + Set out = new HashSet<>(); + for (URI u : uris) out.add(u.getHost()); + return out; + } + + @Test + public void sourceNodeIsAddedEvenWhenAbsentFromDiscoveredList() throws Exception { + // 10.0.0.1 responds but omits itself from its /localnodes list. + ControllableHttpClient http = new ControllableHttpClient("[\"10.0.0.2\",\"10.0.0.3\"]"); + AlternatorConfig config = + AlternatorConfig.builder().withSeedNode(URI.create("http://10.0.0.1:8000")).build(); + AlternatorLiveNodes liveNodes = new AlternatorLiveNodes(config, http); + + liveNodes.updateLiveNodes(); + + Set live = hostsOf(liveNodes.getLiveNodes()); + assertTrue( + "10.0.0.1 (responder) must be added even though it omitted itself", + live.contains("10.0.0.1")); + assertTrue(live.contains("10.0.0.2")); + assertTrue(live.contains("10.0.0.3")); + } + + @Test + public void updateLiveNodesUpdatesLastActivityTime() throws Exception { + ControllableHttpClient http = new ControllableHttpClient("[\"10.0.0.1\"]"); + AlternatorConfig config = + AlternatorConfig.builder().withSeedNode(URI.create("http://10.0.0.1:8000")).build(); + AlternatorLiveNodes liveNodes = new AlternatorLiveNodes(config, http); + + Field field = AlternatorLiveNodes.class.getDeclaredField("lastActivityTime"); + field.setAccessible(true); + AtomicLong lastActivityTime = (AtomicLong) field.get(liveNodes); + + long before = System.currentTimeMillis(); + liveNodes.updateLiveNodes(); + long after = System.currentTimeMillis(); + + long recorded = lastActivityTime.get(); + assertTrue("lastActivityTime must be >= time before the call", recorded >= before); + assertTrue("lastActivityTime must be <= time after the call", recorded <= after); + } + + @Test + @SuppressWarnings("unchecked") + public void drainBufferIs1KBAndReusedOnSameThread() throws Exception { + Field field = AlternatorLiveNodes.class.getDeclaredField("DRAIN_BUFFER"); + field.setAccessible(true); + ThreadLocal drainBuffer = (ThreadLocal) field.get(null); + + byte[] first = drainBuffer.get(); + assertEquals("drain buffer must be 1024 bytes", 1024, first.length); + assertSame( + "drain buffer must be the same instance on repeated calls (no re-alloc)", + first, + drainBuffer.get()); + } + + @Test + @SuppressWarnings("unchecked") + public void drainBufferIsUsedWhenBodyIsConsumedOnNon200Response() throws Exception { + // A non-200 response triggers consumeAndClose(), which reads via DRAIN_BUFFER. + // Verify that the buffer is still intact after the call (not zeroed or replaced). + ControllableHttpClient http = new ControllableHttpClient("error body"); + http.setStatusCode(503); + AlternatorConfig config = + AlternatorConfig.builder().withSeedNode(URI.create("http://10.0.0.1:8000")).build(); + AlternatorLiveNodes liveNodes = new AlternatorLiveNodes(config, http); + + Field field = AlternatorLiveNodes.class.getDeclaredField("DRAIN_BUFFER"); + field.setAccessible(true); + ThreadLocal drainBuffer = (ThreadLocal) field.get(null); + + byte[] bufBefore = drainBuffer.get(); + // updateLiveNodes() gets a 503; consumeAndClose() drains the body via DRAIN_BUFFER. + liveNodes.updateLiveNodes(); + byte[] bufAfter = drainBuffer.get(); + + assertSame( + "DRAIN_BUFFER must be the same instance after consumeAndClose()", bufBefore, bufAfter); + assertEquals("buffer length must remain 1024", 1024, bufAfter.length); + } + + @Test + public void emptyNodeListResponseCausesScopeFallthrough() throws Exception { + // A 200 response with body "[]" yields an empty node list. The refresh must not publish + // it; getNodesForScope() returns an empty list and updateLiveNodes() falls through to the + // next scope (or leaves the live list untouched if no scope remains). + ControllableHttpClient http = new ControllableHttpClient("[]"); + AlternatorConfig config = + AlternatorConfig.builder().withSeedNode(URI.create("http://10.0.0.1:8000")).build(); + AlternatorLiveNodes liveNodes = new AlternatorLiveNodes(config, http); + + liveNodes.updateLiveNodes(); + + // No scope returned nodes → live list falls back to initial seeds. + assertFalse( + "live list must not be empty after all-empty refresh", liveNodes.getLiveNodes().isEmpty()); + } + + @Test + public void getNodesReturnsEmptyListWhen200ResponseHasNoBody() throws Exception { + SdkHttpClient noBodyClient = + new SdkHttpClient() { + @Override + public ExecutableHttpRequest prepareRequest(HttpExecuteRequest request) { + return new ExecutableHttpRequest() { + @Override + public HttpExecuteResponse call() { + return HttpExecuteResponse.builder() + .response(SdkHttpFullResponse.builder().statusCode(200).build()) + // no responseBody() → Optional.empty() + .build(); + } + + @Override + public void abort() {} + }; + } + + @Override + public void close() {} + + @Override + public String clientName() { + return "NoBodyClient"; + } + }; + + AlternatorConfig config = + AlternatorConfig.builder().withSeedNode(URI.create("http://10.0.0.1:8000")).build(); + AlternatorLiveNodes liveNodes = new AlternatorLiveNodes(config, noBodyClient); + + // Must not throw; falls through to seed-restore since no nodes found. + liveNodes.updateLiveNodes(); + assertFalse(liveNodes.getLiveNodes().isEmpty()); + } + + @Test + public void consumeAndCloseAbsorbsReadIOException() throws Exception { + // Stream that throws on read() but abort() is a no-op — covers the IOException catch path. + InputStream throwingRead = + new InputStream() { + @Override + public int read() throws IOException { + throw new IOException("simulated read failure"); + } + + @Override + public int read(byte[] b, int off, int len) throws IOException { + throw new IOException("simulated read failure"); + } + }; + SdkHttpClient client = singleResponseClient(503, AbortableInputStream.create(throwingRead)); + AlternatorLiveNodes liveNodes = + new AlternatorLiveNodes( + AlternatorConfig.builder().withSeedNode(URI.create("http://10.0.0.1:8000")).build(), + client); + liveNodes.updateLiveNodes(); // must not throw + } + + @Test + public void consumeAndCloseAbsorbsAbortExceptionAfterReadFailure() throws Exception { + // Both read() and abort() throw — covers the inner catch(Exception abortEx) path. + InputStream throwingRead = + new InputStream() { + @Override + public int read() throws IOException { + throw new IOException("simulated read failure"); + } + + @Override + public int read(byte[] b, int off, int len) throws IOException { + throw new IOException("simulated read failure"); + } + }; + Abortable throwingAbort = + () -> { + throw new RuntimeException("simulated abort failure"); + }; + SdkHttpClient client = + singleResponseClient(503, AbortableInputStream.create(throwingRead, throwingAbort)); + AlternatorLiveNodes liveNodes = + new AlternatorLiveNodes( + AlternatorConfig.builder().withSeedNode(URI.create("http://10.0.0.1:8000")).build(), + client); + liveNodes.updateLiveNodes(); // must not throw + } + + private static SdkHttpClient singleResponseClient(int statusCode, AbortableInputStream body) { + return new SdkHttpClient() { + @Override + public ExecutableHttpRequest prepareRequest(HttpExecuteRequest request) { + return new ExecutableHttpRequest() { + @Override + public HttpExecuteResponse call() { + return HttpExecuteResponse.builder() + .response(SdkHttpFullResponse.builder().statusCode(statusCode).build()) + .responseBody(body) + .build(); + } + + @Override + public void abort() {} + }; + } + + @Override + public void close() {} + + @Override + public String clientName() { + return "SingleResponseClient"; + } + }; + } + + // Direct parseLocalNodes branch tests — placed here because forkCount=2 in Surefire + // isolates ParseLocalNodesTest to a separate JVM whose JaCoCo data does not merge into + // the main exec file. Tests here share the fork that covers AlternatorLiveNodes. + + @Test + public void parseLocalNodes_whitespaceOnly() throws Exception { + AlternatorLiveNodes ln = + new AlternatorLiveNodes( + AlternatorConfig.builder().withSeedNode(URI.create("http://10.0.0.1:8000")).build()); + assertTrue(ln.parseLocalNodes(" \t\n").isEmpty()); + } + + @Test + public void parseLocalNodes_unclosedString() throws Exception { + // Truncated mid-string; inner while exits when i >= n. + AlternatorLiveNodes ln = + new AlternatorLiveNodes( + AlternatorConfig.builder().withSeedNode(URI.create("http://10.0.0.1:8000")).build()); + List r = ln.parseLocalNodes("[\"host"); + assertEquals(1, r.size()); // "host" is a valid URI hostname + } + + @Test + public void parseLocalNodes_backslashAtEndOfInput() throws Exception { + // ch=='\\' AND i+1 >= n: condition is false, backslash appended as literal char. + AlternatorLiveNodes ln = + new AlternatorLiveNodes( + AlternatorConfig.builder().withSeedNode(URI.create("http://10.0.0.1:8000")).build()); + // "[\"10.0.0.\\" Java literal → ["10.0.0.\ raw: backslash is the last char. + assertTrue(ln.parseLocalNodes("[\"10.0.0.\\").isEmpty()); + } + + @Test + public void parseLocalNodes_leadingSlashHostTriggersPortMismatch() throws Exception { + // host="/10.0.0.1" causes URI() to produce port -1; port check throws URISyntaxException. + AlternatorLiveNodes ln = + new AlternatorLiveNodes( + AlternatorConfig.builder().withSeedNode(URI.create("http://10.0.0.1:8000")).build()); + List r = ln.parseLocalNodes("[\"/10.0.0.1\",\"10.0.0.2\"]"); + assertEquals(1, r.size()); + assertEquals("10.0.0.2", r.get(0).getHost()); + } + + @Test + public void parseLocalNodes_null() throws Exception { + AlternatorLiveNodes ln = + new AlternatorLiveNodes( + AlternatorConfig.builder().withSeedNode(URI.create("http://10.0.0.1:8000")).build()); + assertTrue(ln.parseLocalNodes(null).isEmpty()); + } + + @Test + public void parseLocalNodes_leadingWhitespace() throws Exception { + AlternatorLiveNodes ln = + new AlternatorLiveNodes( + AlternatorConfig.builder().withSeedNode(URI.create("http://10.0.0.1:8000")).build()); + List r = ln.parseLocalNodes(" [ \"10.0.0.1\" ]"); + assertEquals(1, r.size()); + } + + @Test + public void parseLocalNodes_nonArray() throws Exception { + AlternatorLiveNodes ln = + new AlternatorLiveNodes( + AlternatorConfig.builder().withSeedNode(URI.create("http://10.0.0.1:8000")).build()); + assertTrue(ln.parseLocalNodes("{\"host\":\"10.0.0.1\"}").isEmpty()); + } + + @Test + public void parseLocalNodes_unexpectedChar() throws Exception { + AlternatorLiveNodes ln = + new AlternatorLiveNodes( + AlternatorConfig.builder().withSeedNode(URI.create("http://10.0.0.1:8000")).build()); + assertTrue(ln.parseLocalNodes("[invalid]").isEmpty()); + } + + @Test + public void parseLocalNodes_backslashEscape() throws Exception { + AlternatorLiveNodes ln = + new AlternatorLiveNodes( + AlternatorConfig.builder().withSeedNode(URI.create("http://10.0.0.1:8000")).build()); + // \"10.0.0\.1\" — the \. escape resolves to '.' giving host 10.0.0.1 + List r = ln.parseLocalNodes("[\"10.0.0\\.1\"]"); + assertEquals(1, r.size()); + assertEquals("10.0.0.1", r.get(0).getHost()); + } + + @Test + public void parseLocalNodes_emptyStringEntry() throws Exception { + AlternatorLiveNodes ln = + new AlternatorLiveNodes( + AlternatorConfig.builder().withSeedNode(URI.create("http://10.0.0.1:8000")).build()); + List r = ln.parseLocalNodes("[\"\",\"10.0.0.1\"]"); + assertEquals(1, r.size()); + } + + @Test + public void parseLocalNodes_invalidHostCaughtAndSkipped() throws Exception { + AlternatorLiveNodes ln = + new AlternatorLiveNodes( + AlternatorConfig.builder().withSeedNode(URI.create("http://10.0.0.1:8000")).build()); + // "host with spaces" throws URISyntaxException in hostToURI; valid host follows + List r = ln.parseLocalNodes("[\"host with spaces\",\"10.0.0.1\"]"); + assertEquals(1, r.size()); + } + + @Test + public void updateLiveNodes_allNodesDownRestoresSeedList() throws Exception { + ControllableHttpClient http = new ControllableHttpClient("[\"10.0.0.1\",\"10.0.0.2\"]"); + AlternatorConfig config = + AlternatorConfig.builder() + .withSeedHosts(Arrays.asList("10.0.0.1", "10.0.0.2")) + .withScheme("http") + .withPort(8000) + .build(); + AlternatorLiveNodes liveNodes = new AlternatorLiveNodes(config, http); + + http.markDown("10.0.0.1"); + http.markDown("10.0.0.2"); + + try { + liveNodes.updateLiveNodes(); + } catch (IOException ignored) { + } + + // All dead → remaining.isEmpty() == true → seed list restored. + assertEquals( + "seed list must be restored when all nodes are down", 2, liveNodes.getLiveNodes().size()); + } + + @Test + public void hostToURICallsToURLWhenSchemeNotYetValidated() throws Exception { + AlternatorConfig config = + AlternatorConfig.builder().withSeedNode(URI.create("http://10.0.0.1:8000")).build(); + AlternatorLiveNodes liveNodes = + new AlternatorLiveNodes(config, new ControllableHttpClient("[\"10.0.0.2\"]")); + + // Reset the flag to exercise the !schemeAndPortValidated branch in hostToURI(). + Field f = AlternatorLiveNodes.class.getDeclaredField("schemeAndPortValidated"); + f.setAccessible(true); + f.setBoolean(liveNodes, false); + + List result = liveNodes.parseLocalNodes("[\"10.0.0.2\"]"); + assertEquals("uri must be returned even when toURL() validation is re-run", 1, result.size()); + assertEquals("http", result.get(0).getScheme()); + assertEquals(8000, result.get(0).getPort()); + } +} diff --git a/src/test/java/com/scylladb/alternator/queryplan/BasicQueryPlanInterceptorTest.java b/src/test/java/com/scylladb/alternator/queryplan/BasicQueryPlanInterceptorTest.java new file mode 100644 index 0000000..1af2dd4 --- /dev/null +++ b/src/test/java/com/scylladb/alternator/queryplan/BasicQueryPlanInterceptorTest.java @@ -0,0 +1,173 @@ +package com.scylladb.alternator.queryplan; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; + +import com.scylladb.alternator.AlternatorConfig; +import com.scylladb.alternator.internal.AlternatorLiveNodes; +import com.scylladb.alternator.internal.LazyQueryPlan; +import java.net.URI; +import java.util.List; +import java.util.Optional; +import org.junit.Before; +import org.junit.Test; +import software.amazon.awssdk.core.SdkRequest; +import software.amazon.awssdk.core.async.AsyncRequestBody; +import software.amazon.awssdk.core.interceptor.Context; +import software.amazon.awssdk.core.interceptor.ExecutionAttributes; +import software.amazon.awssdk.core.sync.RequestBody; +import software.amazon.awssdk.http.SdkHttpMethod; +import software.amazon.awssdk.http.SdkHttpRequest; +import software.amazon.awssdk.services.dynamodb.model.ListTablesRequest; + +/** + * Unit tests for {@link BasicQueryPlanInterceptor#modifyHttpRequest}. + * + *

Focuses on the changed behaviour: the {@code Connection: keep-alive} header is added only when + * absent. Adding it unconditionally on every retry forces the SDK to reallocate the full header map + * even when the underlying transport has already set it. + */ +public class BasicQueryPlanInterceptorTest { + + private static final URI NODE = URI.create("http://10.0.0.1:8000"); + + private AlternatorLiveNodes liveNodes; + private BasicQueryPlanInterceptor interceptor; + + @Before + public void setUp() { + liveNodes = new AlternatorLiveNodes(AlternatorConfig.builder().withSeedNode(NODE).build()); + interceptor = new BasicQueryPlanInterceptor(liveNodes); + } + + // ------------------------------------------------------------------------- + // Helpers + // ------------------------------------------------------------------------- + + private SdkHttpRequest baseRequest() { + return SdkHttpRequest.builder() + .protocol("http") + .host("placeholder") + .port(9999) + .method(SdkHttpMethod.POST) + .encodedPath("/") + .build(); + } + + private SdkHttpRequest requestWithHeader(String name, String value) { + return baseRequest().toBuilder().putHeader(name, value).build(); + } + + private Context.ModifyHttpRequest ctx(SdkHttpRequest req) { + return new Context.ModifyHttpRequest() { + @Override + public SdkHttpRequest httpRequest() { + return req; + } + + @Override + public Optional requestBody() { + return Optional.empty(); + } + + @Override + public Optional asyncRequestBody() { + return Optional.empty(); + } + + @Override + public SdkRequest request() { + return ListTablesRequest.builder().build(); + } + }; + } + + private ExecutionAttributes attributesWithPlan() { + ExecutionAttributes attrs = new ExecutionAttributes(); + interceptor.beforeExecution(null, attrs); + return attrs; + } + + @Test + public void keepAliveHeaderAddedWhenAbsent() { + ExecutionAttributes attrs = attributesWithPlan(); + SdkHttpRequest result = interceptor.modifyHttpRequest(ctx(baseRequest()), attrs); + + List connection = result.matchingHeaders("Connection"); + assertFalse( + "Connection header must be present when absent from original", connection.isEmpty()); + assertTrue( + "Connection header must contain keep-alive", + connection.stream().anyMatch(v -> v.equalsIgnoreCase("keep-alive"))); + } + + @Test + public void keepAliveHeaderNotDuplicatedWhenAlreadyPresent() { + ExecutionAttributes attrs = attributesWithPlan(); + SdkHttpRequest incoming = requestWithHeader("Connection", "keep-alive"); + SdkHttpRequest result = interceptor.modifyHttpRequest(ctx(incoming), attrs); + + List connection = result.matchingHeaders("Connection"); + long count = connection.stream().filter(v -> v.equalsIgnoreCase("keep-alive")).count(); + assertEquals("keep-alive must appear exactly once, not duplicated", 1, count); + } + + @Test + public void existingConnectionHeaderNotOverriddenWithDifferentValue() { + ExecutionAttributes attrs = attributesWithPlan(); + SdkHttpRequest incoming = requestWithHeader("Connection", "close"); + SdkHttpRequest result = interceptor.modifyHttpRequest(ctx(incoming), attrs); + + // The interceptor should leave the existing Connection header alone, adding nothing. + List connection = result.matchingHeaders("Connection"); + assertEquals("only the original Connection value expected", 1, connection.size()); + assertEquals("close", connection.get(0)); + } + + @Test + public void requestIsRewrittenToLiveNodeHostAndPort() { + ExecutionAttributes attrs = attributesWithPlan(); + SdkHttpRequest result = interceptor.modifyHttpRequest(ctx(baseRequest()), attrs); + + assertEquals("host must be rewritten to live node", NODE.getHost(), result.host()); + assertEquals("port must be rewritten to live node", NODE.getPort(), result.port()); + assertEquals("scheme must be rewritten to live node", NODE.getScheme(), result.protocol()); + } + + @Test + public void noPlanAttributeReturnOriginalRequest() { + SdkHttpRequest original = baseRequest(); + SdkHttpRequest result = interceptor.modifyHttpRequest(ctx(original), new ExecutionAttributes()); + assertSame("original request returned when no plan attribute is present", original, result); + } + + @Test + public void exhaustedPlanReturnsOriginalRequest() { + ExecutionAttributes attrs = attributesWithPlan(); + LazyQueryPlan plan = attrs.getAttribute(BasicQueryPlanInterceptor.QUERY_PLAN); + while (plan.hasNext()) { + plan.next(); + } + SdkHttpRequest original = baseRequest(); + SdkHttpRequest result = interceptor.modifyHttpRequest(ctx(original), attrs); + assertSame("original request returned when plan is exhausted", original, result); + } + + // ------------------------------------------------------------------------- + // beforeExecution installs a plan + // ------------------------------------------------------------------------- + + @Test + public void beforeExecutionInstallsQueryPlan() { + ExecutionAttributes attrs = new ExecutionAttributes(); + interceptor.beforeExecution(null, attrs); + + com.scylladb.alternator.internal.LazyQueryPlan plan = + attrs.getAttribute(BasicQueryPlanInterceptor.QUERY_PLAN); + assertNotNull("beforeExecution must install a query plan", plan); + assertTrue("installed plan must have at least one node", plan.hasNext()); + } +}