From 64aa496d0946cec0a37df1f4f2060f296036b47d Mon Sep 17 00:00:00 2001 From: Dmitry Kropachev Date: Tue, 21 Jul 2026 12:02:07 -0400 Subject: [PATCH] refactor: parse localnodes responses with streaming parser --- .../internal/AlternatorLiveNodes.java | 45 +---- .../internal/LocalNodesResponseParser.java | 165 ++++++++++++++++++ ...ternatorLiveNodesClusterDiscoveryTest.java | 22 +++ .../LocalNodesResponseParserTest.java | 66 +++++++ 4 files changed, 261 insertions(+), 37 deletions(-) create mode 100644 src/main/java/com/scylladb/alternator/internal/LocalNodesResponseParser.java create mode 100644 src/test/java/com/scylladb/alternator/internal/LocalNodesResponseParserTest.java diff --git a/src/main/java/com/scylladb/alternator/internal/AlternatorLiveNodes.java b/src/main/java/com/scylladb/alternator/internal/AlternatorLiveNodes.java index 99b1865..979f5be 100644 --- a/src/main/java/com/scylladb/alternator/internal/AlternatorLiveNodes.java +++ b/src/main/java/com/scylladb/alternator/internal/AlternatorLiveNodes.java @@ -39,8 +39,6 @@ public class AlternatorLiveNodes extends Thread { private static final long DEFAULT_SHUTDOWN_TIMEOUT_MS = 5_000; - private final String alternatorScheme; - private final int alternatorPort; private final AtomicReference> liveNodes; private final List initialNodes; private final AtomicInteger nextLiveNodeIndex; @@ -50,6 +48,7 @@ public class AlternatorLiveNodes extends Thread { private final SdkHttpClient pollingHttpClient; private final boolean ownsPollingClient; private final AtomicLong lastActivityTime = new AtomicLong(0); + private final LocalNodesResponseParser localNodesResponseParser; private static Logger logger = Logger.getLogger(AlternatorLiveNodes.class.getName()); @@ -359,8 +358,8 @@ private AlternatorLiveNodes( if (seedHosts == null || seedHosts.isEmpty()) { throw new RuntimeException("config must contain at least one seed host"); } - this.alternatorScheme = config.getScheme(); - this.alternatorPort = config.getPort(); + this.localNodesResponseParser = + new LocalNodesResponseParser(config.getScheme(), config.getPort()); this.initialNodes = hostsToUris(seedHosts); this.liveNodes = new AtomicReference<>(); this.nextLiveNodeIndex = new AtomicInteger(0); @@ -450,7 +449,7 @@ public ValidationError(String message, Throwable cause) { private void validateConfig() throws ValidationError { try { - // Make sure that `alternatorScheme` and `alternatorPort` are correct values + // Make sure that the configured scheme and port are valid values. this.hostToURI("1.1.1.1"); } catch (MalformedURLException | URISyntaxException e) { throw new ValidationError("failed to validate configuration", e); @@ -458,10 +457,7 @@ 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(); - return uri; + return localNodesResponseParser.hostToURI(host); } private List hostsToUris(List hosts) { @@ -695,37 +691,12 @@ 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("]")) { + try { + return localNodesResponseParser.parse(responseStr); + } catch (LocalNodesResponseParser.InvalidLocalNodesResponseException e) { logger.log(Level.WARNING, "Malformed /localnodes response: " + responseStr); return Collections.emptyList(); } - - String responseBody = responseStr.substring(1, responseStr.length() - 1).trim(); - if (responseBody.isEmpty()) { - return Collections.emptyList(); - } - - String[] list = responseBody.split(","); - List newHosts = new ArrayList<>(); - for (String host : list) { - host = host.trim(); - if (host.isEmpty()) { - continue; - } - if (host.length() < 2 || !host.startsWith("\"") || !host.endsWith("\"")) { - logger.log(Level.WARNING, "Malformed host entry in /localnodes response: " + host); - continue; - } - host = host.substring(1, host.length() - 1); - try { - newHosts.add(this.hostToURI(host)); - } catch (URISyntaxException | MalformedURLException e) { - logger.log(Level.WARNING, "Invalid host: " + host, e); - } - } - return newHosts; } /** diff --git a/src/main/java/com/scylladb/alternator/internal/LocalNodesResponseParser.java b/src/main/java/com/scylladb/alternator/internal/LocalNodesResponseParser.java new file mode 100644 index 0000000..097e51b --- /dev/null +++ b/src/main/java/com/scylladb/alternator/internal/LocalNodesResponseParser.java @@ -0,0 +1,165 @@ +package com.scylladb.alternator.internal; + +import java.io.IOException; +import java.net.MalformedURLException; +import java.net.URI; +import java.net.URISyntaxException; +import java.util.ArrayList; +import java.util.List; +import java.util.logging.Level; +import java.util.logging.Logger; + +final class LocalNodesResponseParser { + private final String alternatorScheme; + private final int alternatorPort; + + private static final Logger logger = Logger.getLogger(LocalNodesResponseParser.class.getName()); + + LocalNodesResponseParser(String alternatorScheme, int alternatorPort) { + this.alternatorScheme = alternatorScheme; + this.alternatorPort = alternatorPort; + } + + List parse(String responseBody) throws InvalidLocalNodesResponseException { + List nodes = new ArrayList<>(); + for (String host : new JsonStringArrayParser(responseBody).parse()) { + try { + nodes.add(hostToURI(host)); + } catch (URISyntaxException | MalformedURLException e) { + logger.log(Level.WARNING, "Invalid host: " + host, e); + } + } + return nodes; + } + + URI hostToURI(String host) throws URISyntaxException, MalformedURLException { + URI uri = new URI(alternatorScheme, null, host, alternatorPort, null, null, null); + uri.toURL(); + return uri; + } + + static class InvalidLocalNodesResponseException extends IOException { + private InvalidLocalNodesResponseException(String message) { + super(message); + } + } + + private static final class JsonStringArrayParser { + private final String body; + private int pos = 0; + + private JsonStringArrayParser(String body) { + this.body = body != null ? body : ""; + } + + private List parse() throws InvalidLocalNodesResponseException { + List values = new ArrayList<>(); + expect('['); + skipWhitespace(); + if (peek(']')) { + pos++; + expectEnd(); + return values; + } + + while (true) { + values.add(parseString()); + skipWhitespace(); + if (peek(']')) { + pos++; + expectEnd(); + return values; + } + expect(','); + } + } + + private String parseString() throws InvalidLocalNodesResponseException { + expect('"'); + StringBuilder value = new StringBuilder(); + while (pos < body.length()) { + char ch = body.charAt(pos++); + if (ch == '"') { + return value.toString(); + } + if (ch != '\\') { + value.append(ch); + continue; + } + value.append(parseEscapedCharacter()); + } + throw invalid("unterminated /localnodes JSON response"); + } + + private char parseEscapedCharacter() throws InvalidLocalNodesResponseException { + if (pos >= body.length()) { + throw invalid("invalid escape in /localnodes JSON response"); + } + char escaped = body.charAt(pos++); + switch (escaped) { + case '"': + case '\\': + case '/': + return escaped; + case 'b': + return '\b'; + case 'f': + return '\f'; + case 'n': + return '\n'; + case 'r': + return '\r'; + case 't': + return '\t'; + case 'u': + return parseUnicodeEscape(); + default: + throw invalid("unsupported escape in /localnodes JSON response"); + } + } + + private char parseUnicodeEscape() throws InvalidLocalNodesResponseException { + if (pos + 4 > body.length()) { + throw invalid("invalid unicode escape in /localnodes JSON response"); + } + int codePoint = 0; + for (int i = 0; i < 4; i++) { + int digit = Character.digit(body.charAt(pos++), 16); + if (digit < 0) { + throw invalid("invalid unicode escape in /localnodes JSON response"); + } + codePoint = (codePoint << 4) + digit; + } + return (char) codePoint; + } + + private void expect(char expected) throws InvalidLocalNodesResponseException { + skipWhitespace(); + if (pos >= body.length() || body.charAt(pos) != expected) { + throw invalid("invalid /localnodes JSON response"); + } + pos++; + } + + private void expectEnd() throws InvalidLocalNodesResponseException { + skipWhitespace(); + if (pos != body.length()) { + throw invalid("invalid trailing data in /localnodes JSON response"); + } + } + + private boolean peek(char expected) { + return pos < body.length() && body.charAt(pos) == expected; + } + + private void skipWhitespace() { + while (pos < body.length() && Character.isWhitespace(body.charAt(pos))) { + pos++; + } + } + + private InvalidLocalNodesResponseException invalid(String message) { + return new InvalidLocalNodesResponseException(message); + } + } +} diff --git a/src/test/java/com/scylladb/alternator/internal/AlternatorLiveNodesClusterDiscoveryTest.java b/src/test/java/com/scylladb/alternator/internal/AlternatorLiveNodesClusterDiscoveryTest.java index aacc0ba..2bf4b63 100644 --- a/src/test/java/com/scylladb/alternator/internal/AlternatorLiveNodesClusterDiscoveryTest.java +++ b/src/test/java/com/scylladb/alternator/internal/AlternatorLiveNodesClusterDiscoveryTest.java @@ -272,6 +272,28 @@ public void testScopedLocalNodesQueryValuesAreEncodedOnce() throws Exception { assertFalse(encodedQuery.contains("%25")); } + @Test + public void testLocalNodesParserHandlesWhitespaceAndEscapedStrings() throws Exception { + Map responses = new HashMap<>(); + responses.put("seed.example.com", " [ \"node1.example.com\" , \"node\\u0032.example.com\" ] "); + DiscoveryHttpClient httpClient = new DiscoveryHttpClient(responses); + + AlternatorConfig config = + AlternatorConfig.builder() + .withSeedHost("seed.example.com") + .withScheme("http") + .withPort(8000) + .withRoutingScope(ClusterScope.create()) + .build(); + + AlternatorLiveNodes liveNodes = new AlternatorLiveNodes(config, httpClient); + liveNodes.updateLiveNodes(); + + assertEquals( + new LinkedHashSet<>(Arrays.asList("node1.example.com", "node2.example.com")), + hostSet(liveNodes.getLiveNodes())); + } + @Test public void testCustomScopeLocalNodesQueryValuesAreEncodedWhenNeeded() throws Exception { Map responses = new HashMap<>(); diff --git a/src/test/java/com/scylladb/alternator/internal/LocalNodesResponseParserTest.java b/src/test/java/com/scylladb/alternator/internal/LocalNodesResponseParserTest.java new file mode 100644 index 0000000..d7b039c --- /dev/null +++ b/src/test/java/com/scylladb/alternator/internal/LocalNodesResponseParserTest.java @@ -0,0 +1,66 @@ +package com.scylladb.alternator.internal; + +import static org.junit.Assert.*; + +import java.net.URI; +import java.util.List; +import org.junit.Test; + +public class LocalNodesResponseParserTest { + private final LocalNodesResponseParser parser = new LocalNodesResponseParser("http", 8000); + + @Test + public void parsesEmptyArrays() throws Exception { + assertTrue(parser.parse("[]").isEmpty()); + assertTrue(parser.parse(" [ ] ").isEmpty()); + } + + @Test + public void parsesValidMultiHostResponses() throws Exception { + List nodes = parser.parse("[\"node1.example.com\",\"node2.example.com\"]"); + + assertHosts(nodes, "node1.example.com", "node2.example.com"); + } + + @Test + public void parsesEscapedHostStrings() throws Exception { + List nodes = + parser.parse(" [ \"node\\u0031.example.com\" , \"node\\u0032.example.com\" ] "); + + assertHosts(nodes, "node1.example.com", "node2.example.com"); + } + + @Test + public void skipsInvalidHostEntries() throws Exception { + List nodes = parser.parse("[\"node1.example.com\",\"bad host\",\"node2.example.com\"]"); + + assertHosts(nodes, "node1.example.com", "node2.example.com"); + } + + @Test + public void rejectsMalformedBodies() throws Exception { + assertMalformed(""); + assertMalformed("not-json"); + assertMalformed("["); + assertMalformed("[bad]"); + assertMalformed("[\"node1.example.com\",bad]"); + } + + private void assertMalformed(String body) throws Exception { + try { + parser.parse(body); + fail("Expected malformed /localnodes body: " + body); + } catch (LocalNodesResponseParser.InvalidLocalNodesResponseException e) { + // expected + } + } + + private static void assertHosts(List nodes, String... expectedHosts) { + assertEquals(expectedHosts.length, nodes.size()); + for (int i = 0; i < expectedHosts.length; i++) { + assertEquals(expectedHosts[i], nodes.get(i).getHost()); + assertEquals("http", nodes.get(i).getScheme()); + assertEquals(8000, nodes.get(i).getPort()); + } + } +}