diff --git a/src/main/java/com/scylladb/alternator/internal/AlternatorLiveNodes.java b/src/main/java/com/scylladb/alternator/internal/AlternatorLiveNodes.java index 0a389e2..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,11 +546,22 @@ private static String streamToString(InputStream stream) throws IOException { } void updateLiveNodes() throws IOException { + // 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; + // 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 +583,34 @@ 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, that's 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 +629,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); @@ -638,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); @@ -646,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 } @@ -750,7 +845,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 " @@ -835,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 new file mode 100644 index 0000000..5c75834 --- /dev/null +++ b/src/test/java/com/scylladb/alternator/internal/AlternatorLiveNodesDeadSeedTest.java @@ -0,0 +1,202 @@ +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; + } +} 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()); + } +}