This is an automated email from the ASF dual-hosted git repository. mmodzelewski pushed a commit to branch java-leader-aware in repository https://gitbox.apache.org/repos/asf/iggy.git
commit da553e2cda1eeab2b70ef5267892e8ccf9f43bbb Author: Maciej Modzelewski <[email protected]> AuthorDate: Thu Jul 23 21:17:00 2026 +0200 feat(java): add cluster metadata and leader redirection to TCP client Java clients connected to a VSR follower had no way to discover the cluster leader, so operations against a multi-node cluster failed until the user manually pointed the client at the right node. Add the GET_CLUSTER_METADATA command with cluster roster models and a binary deserializer. After login the async TCP client fetches the roster and, when a healthy leader with an enabled TCP endpoint lives elsewhere, reconnects to it and replays the login. Redirections are capped at 3 consecutive hops so two nodes both claiming leadership cannot ping-pong a client, and every failure on the redirect path is non-fatal: the client stays on the current node. Mirrors the Rust and Go SDKs. Covered by a leader_redirection BDD scenario wired into the cluster docker-compose setup. --- bdd/docker-compose.cluster.yml | 3 + bdd/docker-compose.yml | 6 +- .../apache/iggy/bdd/LeaderRedirectionSteps.java | 274 +++++++++++++++++ foreign/java/gradle.properties | 2 +- .../SystemClient.java => ConnectionInfo.java} | 37 +-- .../org/apache/iggy/client/async/SystemClient.java | 8 + .../iggy/client/async/tcp/AsyncIggyTcpClient.java | 199 +++++++++++-- .../iggy/client/async/tcp/AsyncTcpConnection.java | 73 +++-- .../client/async/tcp/ConsumerGroupsTcpClient.java | 23 +- .../client/async/tcp/ConsumerOffsetsTcpClient.java | 15 +- .../iggy/client/async/tcp/LeaderAwareness.java | 282 ++++++++++++++++++ .../client/async/tcp/LoginRedirectionHook.java | 47 +++ .../iggy/client/async/tcp/MessagesTcpClient.java | 15 +- .../iggy/client/async/tcp/PartitionsTcpClient.java | 27 +- .../async/tcp/PersonalAccessTokensTcpClient.java | 32 +- .../iggy/client/async/tcp/StreamsTcpClient.java | 21 +- .../iggy/client/async/tcp/SystemTcpClient.java | 53 +++- .../iggy/client/async/tcp/TopicsTcpClient.java | 25 +- .../iggy/client/async/tcp/UsersTcpClient.java | 42 ++- .../apache/iggy/client/blocking/SystemClient.java | 3 + .../client/blocking/http/SystemHttpClient.java | 8 + .../iggy/client/blocking/tcp/IggyTcpClient.java | 12 + .../iggy/client/blocking/tcp/SystemTcpClient.java | 6 + .../ClusterMetadata.java} | 28 +- .../SystemClient.java => cluster/ClusterNode.java} | 31 +- .../ClusterNodeRole.java} | 47 +-- .../org/apache/iggy/cluster/ClusterNodeStatus.java | 58 ++++ .../TransportEndpoints.java} | 30 +- .../package-info.java} | 22 +- .../org/apache/iggy/serde/BytesDeserializer.java | 47 +++ .../java/org/apache/iggy/serde/CommandCode.java | 1 + .../client/async/tcp/IggyResponseHandlerTest.java | 60 ++++ .../iggy/client/async/tcp/LeaderAwarenessTest.java | 328 +++++++++++++++++++++ .../iggy/client/blocking/SystemClientBaseTest.java | 18 ++ .../client/blocking/tcp/SystemTcpClientTest.java | 10 + .../apache/iggy/serde/BytesDeserializerTest.java | 177 +++++++++++ scripts/run-bdd-tests.sh | 2 +- 37 files changed, 1830 insertions(+), 242 deletions(-) diff --git a/bdd/docker-compose.cluster.yml b/bdd/docker-compose.cluster.yml index 7f6279621..063daf0db 100644 --- a/bdd/docker-compose.cluster.yml +++ b/bdd/docker-compose.cluster.yml @@ -120,6 +120,9 @@ services: csharp-bdd: <<: *cluster-bdd-deps + java-bdd: + <<: *cluster-bdd-deps + volumes: iggy_leader_data: iggy_follower_data: diff --git a/bdd/docker-compose.yml b/bdd/docker-compose.yml index 8a103db89..3069546d5 100644 --- a/bdd/docker-compose.yml +++ b/bdd/docker-compose.yml @@ -170,14 +170,16 @@ services: - BDD_FEATURE=${BDD_FEATURE:-all} volumes: - ./scenarios/basic_messaging.feature:/app/features/basic_messaging.feature + - ./scenarios/leader_redirection.feature:/app/features/leader_redirection.feature - ./scenarios/raw_command.feature:/app/features/raw_command.feature command: - sh - -c - | case "$$BDD_FEATURE" in - basic_messaging) export CUCUMBER_FILTER_TAGS='@basic-messaging' ;; - raw_command) export CUCUMBER_FILTER_TAGS='@raw-command' ;; + basic_messaging) export CUCUMBER_FILTER_TAGS='@basic-messaging' ;; + leader_redirection) export CUCUMBER_FILTER_TAGS='@requires-leader-awareness' ;; + raw_command) export CUCUMBER_FILTER_TAGS='@raw-command' ;; esac gradle --no-daemon test networks: diff --git a/bdd/java/src/test/java/org/apache/iggy/bdd/LeaderRedirectionSteps.java b/bdd/java/src/test/java/org/apache/iggy/bdd/LeaderRedirectionSteps.java new file mode 100644 index 000000000..5dc9778b7 --- /dev/null +++ b/bdd/java/src/test/java/org/apache/iggy/bdd/LeaderRedirectionSteps.java @@ -0,0 +1,274 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iggy.bdd; + +import io.cucumber.java.After; +import io.cucumber.java.en.Given; +import io.cucumber.java.en.Then; +import io.cucumber.java.en.When; +import org.apache.iggy.client.blocking.tcp.IggyTcpClient; +import org.apache.iggy.cluster.ClusterNode; +import org.apache.iggy.cluster.ClusterNodeRole; +import org.apache.iggy.cluster.ClusterNodeStatus; +import org.apache.iggy.exception.IggyException; +import org.apache.iggy.stream.StreamDetails; + +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.Optional; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class LeaderRedirectionSteps { + + private static final String MAIN_CLIENT = "main"; + + private final Map<String, IggyTcpClient> clients = new LinkedHashMap<>(); + private boolean redirectionOccurred; + private Long lastStreamId; + + @After + public void closeClients() { + clients.values().forEach(IggyTcpClient::close); + clients.clear(); + } + + // ---------- Background ---------- + + @Given("I have cluster configuration enabled with {int} nodes") + public void clusterConfigurationEnabled(int nodeCount) { + assertTrue(nodeCount > 0, "Cluster must have at least one node"); + } + + @Given("node {int} is configured on port {int}") + public void nodeConfiguredOnPort(int nodeId, int port) { + assertFalse(addressForPort(port).isBlank(), "Node " + nodeId + " address should be configured"); + } + + // ---------- Server start steps (validate addresses are configured) ---------- + + @Given("^I start server (\\d+) on port (\\d+) as (leader|follower)$") + public void startServerAs(int nodeId, int port, String role) { + assertAddressMatchesPort(addressForRole(role), port, role + " address"); + } + + @Given("I start a single server on port {int} without clustering enabled") + public void startSingleServer(int port) { + assertAddressMatchesPort(singleServerAddress(), port, "single-server address"); + } + + // ---------- Client creation ---------- + + @When("^I create a client connecting to (follower|leader) on port (\\d+)$") + public void createClientConnectingToRole(String role, int port) { + String address = addressForRole(role); + assertAddressMatchesPort(address, port, role + " address"); + createAndConnectClient(MAIN_CLIENT, address); + } + + @When("I create a client connecting directly to leader on port {int}") + public void createClientConnectingDirectlyToLeader(int port) { + String address = addressForRole("leader"); + assertAddressMatchesPort(address, port, "leader address"); + createAndConnectClient(MAIN_CLIENT, address); + redirectionOccurred = false; + } + + @When("I create a client connecting to port {int}") + public void createClientConnectingToPort(int port) { + createAndConnectClient(MAIN_CLIENT, addressForPort(port)); + } + + @When("^I create client ([A-Z]) connecting to port (\\d+)$") + public void createNamedClientConnectingToPort(String clientName, int port) { + createAndConnectClient(clientName, addressForPort(port)); + } + + // ---------- Auth ---------- + + @When("I authenticate as root user") + public void authenticateAsRootUser() { + authenticateAllClients(); + } + + @When("both clients authenticate as root user") + public void bothClientsAuthenticateAsRootUser() { + authenticateAllClients(); + } + + // ---------- Stream operations ---------- + + @When("I create a stream named {string}") + public void createStreamNamed(String streamName) { + StreamDetails stream = client(MAIN_CLIENT).streams().createStream(streamName); + assertNotNull(stream, "Stream should be created"); + lastStreamId = stream.id(); + } + + @Then("the stream should be created successfully on the leader") + public void streamCreatedSuccessfullyOnLeader() { + assertNotNull(lastStreamId, "Stream should have been created on the leader"); + } + + // ---------- Connection / redirection assertions ---------- + + @Then("the client should automatically redirect to leader on port {int}") + public void clientAutomaticallyRedirectsToLeader(int expectedPort) { + verifyClientPort(MAIN_CLIENT, expectedPort); + } + + @Then("^client ([A-Z]) should stay connected to port (\\d+)$") + public void namedClientStaysConnectedToPort(String clientName, int expectedPort) { + verifyClientPort(clientName, expectedPort); + } + + @Then("^client ([A-Z]) should redirect to port (\\d+)$") + public void namedClientRedirectsToPort(String clientName, int expectedPort) { + verifyClientPort(clientName, expectedPort); + } + + @Then("the client should not perform any redirection") + public void clientDoesNotRedirect() { + assertFalse(redirectionOccurred, "No redirection should occur when connecting directly to the leader"); + } + + @Then("the connection should remain on port {int}") + public void connectionRemainsOnPort(int expectedPort) { + verifyClientPort(MAIN_CLIENT, expectedPort); + assertFalse(redirectionOccurred, "Connection should not have been redirected"); + } + + @Then("the client should connect successfully without redirection") + public void clientConnectsWithoutRedirection() { + client(MAIN_CLIENT).system().ping(); + assertFalse(redirectionOccurred, "No redirection should occur without clustering"); + } + + @Then("both clients should be using the same server") + public void bothClientsUseTheSameServer() { + IggyTcpClient clientA = client("A"); + IggyTcpClient clientB = client("B"); + + assertEquals( + clientA.getConnectionInfo().serverAddress(), + clientB.getConnectionInfo().serverAddress(), + "Both clients should be connected to the same server"); + + clientA.system().ping(); + clientB.system().ping(); + + Optional<ClusterNode> leaderA = leaderFromMetadata(clientA); + Optional<ClusterNode> leaderB = leaderFromMetadata(clientB); + if (leaderA.isPresent() && leaderB.isPresent()) { + assertEquals( + leaderA.get().ip() + ":" + leaderA.get().endpoints().tcp(), + leaderB.get().ip() + ":" + leaderB.get().endpoints().tcp(), + "Both clients should see the same leader"); + } + } + + // ---------- Helpers ---------- + + private void createAndConnectClient(String name, String address) { + String[] hostPort = address.split(":", 2); + IggyTcpClient client = IggyTcpClient.builder() + .host(hostPort[0]) + .port(Integer.parseInt(hostPort[1])) + .build(); + client.connect(); + clients.put(name, client); + } + + private void authenticateAllClients() { + String username = getenvOrDefault("IGGY_ROOT_USERNAME", "iggy"); + String password = getenvOrDefault("IGGY_ROOT_PASSWORD", "iggy"); + for (IggyTcpClient client : clients.values()) { + String initialAddress = client.getConnectionInfo().serverAddress(); + client.users().login(username, password); + if (!initialAddress.equals(client.getConnectionInfo().serverAddress())) { + redirectionOccurred = true; + } + } + } + + private void verifyClientPort(String clientName, int expectedPort) { + IggyTcpClient client = client(clientName); + assertEquals( + expectedPort, + client.getConnectionInfo().port(), + "Client " + clientName + " should be connected to port " + expectedPort); + client.system().ping(); + } + + private IggyTcpClient client(String name) { + IggyTcpClient client = clients.get(name); + if (client == null) { + throw new IllegalStateException("Client " + name + " should exist"); + } + return client; + } + + /** + * The leader per the roster, or empty when clustering is unavailable + * (a server too old for the command, or metadata otherwise rejected). + */ + private static Optional<ClusterNode> leaderFromMetadata(IggyTcpClient client) { + try { + return client.system().getClusterMetadata().nodes().stream() + .filter(node -> node.role() == ClusterNodeRole.Leader && node.status() == ClusterNodeStatus.Healthy) + .findFirst(); + } catch (IggyException error) { + return Optional.empty(); + } + } + + private static void assertAddressMatchesPort(String address, int port, String description) { + assertTrue(address.endsWith(":" + port), description + " " + address + " should use port " + port); + } + + private static String addressForRole(String role) { + return switch (role) { + case "leader" -> getenvOrDefault("IGGY_TCP_ADDRESS_LEADER", "127.0.0.1:8091"); + case "follower" -> getenvOrDefault("IGGY_TCP_ADDRESS_FOLLOWER", "127.0.0.1:8092"); + default -> throw new IllegalArgumentException("Unknown role: " + role); + }; + } + + private static String addressForPort(int port) { + return switch (port) { + case 8090 -> singleServerAddress(); + case 8091 -> addressForRole("leader"); + case 8092 -> addressForRole("follower"); + default -> throw new IllegalArgumentException("Unknown port: " + port); + }; + } + + private static String singleServerAddress() { + return getenvOrDefault("IGGY_TCP_ADDRESS", "127.0.0.1:8090"); + } + + private static String getenvOrDefault(String key, String defaultValue) { + String value = System.getenv(key); + return value == null || value.isBlank() ? defaultValue : value; + } +} diff --git a/foreign/java/gradle.properties b/foreign/java/gradle.properties index 5339adf06..f9eb048ae 100644 --- a/foreign/java/gradle.properties +++ b/foreign/java/gradle.properties @@ -15,5 +15,5 @@ # specific language governing permissions and limitations # under the License. -version=0.8.2-SNAPSHOT +version=0.8.3-SNAPSHOT group=org.apache.iggy diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/SystemClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/ConnectionInfo.java similarity index 60% copy from foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/SystemClient.java copy to foreign/java/java-sdk/src/main/java/org/apache/iggy/client/ConnectionInfo.java index ecf29e9aa..7042c5fce 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/SystemClient.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/ConnectionInfo.java @@ -17,23 +17,24 @@ * under the License. */ -package org.apache.iggy.client.blocking; +package org.apache.iggy.client; -import org.apache.iggy.system.ClientInfo; -import org.apache.iggy.system.ClientInfoDetails; -import org.apache.iggy.system.Stats; - -import java.util.List; - -public interface SystemClient { - - Stats getStats(); - - ClientInfoDetails getMe(); - - ClientInfoDetails getClient(Long clientId); - - List<ClientInfo> getClients(); - - String ping(); +/** + * The server address a client currently targets. Leader redirection can + * change it after login, so it may differ from the address the client was + * built with. + * + * @param host the server hostname or IP address + * @param port the server port + */ +public record ConnectionInfo(String host, int port) { + + /** + * Returns the address in {@code host:port} form. + * + * @return the server address + */ + public String serverAddress() { + return host + ":" + port; + } } diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/SystemClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/SystemClient.java index f5587142d..2ac3f0b3a 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/SystemClient.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/SystemClient.java @@ -19,6 +19,7 @@ package org.apache.iggy.client.async; +import org.apache.iggy.cluster.ClusterMetadata; import org.apache.iggy.system.ClientInfo; import org.apache.iggy.system.ClientInfoDetails; import org.apache.iggy.system.Stats; @@ -38,6 +39,13 @@ public interface SystemClient { */ CompletableFuture<Stats> getStats(); + /** + * Gets cluster metadata asynchronously. + * + * @return A CompletableFuture containing the cluster roster + */ + CompletableFuture<ClusterMetadata> getClusterMetadata(); + /** * Gets information about the current client asynchronously. * diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClient.java index 0af9a619c..5a78acdba 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClient.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClient.java @@ -21,6 +21,7 @@ package org.apache.iggy.client.async.tcp; import io.netty.buffer.Unpooled; import org.apache.iggy.IggyVersion; +import org.apache.iggy.client.ConnectionInfo; import org.apache.iggy.client.async.ConsumerGroupsClient; import org.apache.iggy.client.async.ConsumerOffsetsClient; import org.apache.iggy.client.async.MessagesClient; @@ -30,7 +31,8 @@ import org.apache.iggy.client.async.StreamsClient; import org.apache.iggy.client.async.SystemClient; import org.apache.iggy.client.async.TopicsClient; import org.apache.iggy.client.async.UsersClient; -import org.apache.iggy.client.async.tcp.AsyncTcpConnection.TCPConnectionPoolConfig; +import org.apache.iggy.client.async.tcp.AsyncTcpConnection.TcpConnectionPoolConfig; +import org.apache.iggy.client.async.tcp.LeaderAwareness.LeaderRedirectionState; import org.apache.iggy.config.RetryPolicy; import org.apache.iggy.exception.IggyMissingCredentialsException; import org.apache.iggy.exception.IggyNotConnectedException; @@ -45,6 +47,9 @@ import java.time.Duration; import java.util.Objects; import java.util.Optional; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Function; +import java.util.function.Supplier; /** * Async TCP client for Apache Iggy message streaming, built on Netty. @@ -101,8 +106,6 @@ public class AsyncIggyTcpClient { private static final int INVALID_COMMAND_ERROR_CODE = 3; private static final Logger log = LoggerFactory.getLogger(AsyncIggyTcpClient.class); - private final String host; - private final int port; private final Optional<String> username; private final Optional<String> password; private final Optional<Duration> connectionTimeout; @@ -112,7 +115,12 @@ public class AsyncIggyTcpClient { private final Optional<RetryPolicy> retryPolicy; private final boolean enableTls; private final Optional<File> tlsCertificate; - private AsyncTcpConnection connection; + private final TcpConnectionPoolConfig poolConfig; + private final AtomicReference<AsyncTcpConnection> connection = new AtomicReference<>(); + private final AtomicReference<CompletableFuture<Void>> redirectChain = + new AtomicReference<>(CompletableFuture.completedFuture(null)); + private volatile ConnectionInfo connectionInfo; + private volatile boolean closed; private MessagesClient messagesClient; private ConsumerGroupsClient consumerGroupsClient; private ConsumerOffsetsClient consumerOffsetsClient; @@ -148,8 +156,7 @@ public class AsyncIggyTcpClient { RetryPolicy retryPolicy, boolean enableTls, Optional<File> tlsCertificate) { - this.host = host; - this.port = port; + this.connectionInfo = new ConnectionInfo(host, port); this.username = Optional.ofNullable(username); this.password = Optional.ofNullable(password); this.connectionTimeout = Optional.ofNullable(connectionTimeout); @@ -159,6 +166,11 @@ public class AsyncIggyTcpClient { this.retryPolicy = Optional.ofNullable(retryPolicy); this.enableTls = enableTls; this.tlsCertificate = tlsCertificate; + + var poolConfigBuilder = TcpConnectionPoolConfig.builder(); + this.connectionPoolSize.ifPresent(poolConfigBuilder::setMaxConnections); + this.acquireTimeout.ifPresent(timeout -> poolConfigBuilder.setAcquireTimeoutMillis(timeout.toMillis())); + this.poolConfig = poolConfigBuilder.build(); } /** @@ -180,22 +192,26 @@ public class AsyncIggyTcpClient { * @return a {@link CompletableFuture} that completes when the connection is established */ public CompletableFuture<Void> connect() { - TCPConnectionPoolConfig.Builder poolConfigBuilder = new TCPConnectionPoolConfig.Builder(); - connectionPoolSize.ifPresent(poolConfigBuilder::setMaxConnections); - acquireTimeout.ifPresent(timeout -> poolConfigBuilder.setAcquireTimeoutMillis(timeout.toMillis())); - TCPConnectionPoolConfig poolConfig = poolConfigBuilder.build(); - connection = new AsyncTcpConnection(host, port, enableTls, tlsCertificate, poolConfig, connectionTimeout); - return connection.connect().thenRun(() -> { - log.debug("Connected to {}:{} | {}", host, port, IggyVersion.getInstance()); - messagesClient = new MessagesTcpClient(connection); - consumerGroupsClient = new ConsumerGroupsTcpClient(connection); - consumerOffsetsClient = new ConsumerOffsetsTcpClient(connection); - streamsClient = new StreamsTcpClient(connection); - topicsClient = new TopicsTcpClient(connection); - usersClient = new UsersTcpClient(connection); - systemClient = new SystemTcpClient(connection); - personalAccessTokensClient = new PersonalAccessTokensTcpClient(connection); - partitionsClient = new PartitionsTcpClient(connection); + closed = false; + ConnectionInfo target = connectionInfo; + AsyncTcpConnection newConnection = openConnection(target); + AsyncTcpConnection previousConnection = connection.getAndSet(newConnection); + if (previousConnection != null) { + previousConnection.close(); + } + Supplier<AsyncTcpConnection> currentConnection = connection::get; + return newConnection.connect().thenRun(() -> { + log.debug("Connected to {} | {}", target.serverAddress(), IggyVersion.getInstance()); + messagesClient = new MessagesTcpClient(currentConnection); + consumerGroupsClient = new ConsumerGroupsTcpClient(currentConnection); + consumerOffsetsClient = new ConsumerOffsetsTcpClient(currentConnection); + streamsClient = new StreamsTcpClient(currentConnection); + topicsClient = new TopicsTcpClient(currentConnection); + usersClient = new UsersTcpClient(currentConnection, this::checkLeaderAndRedirect); + systemClient = new SystemTcpClient(currentConnection); + personalAccessTokensClient = + new PersonalAccessTokensTcpClient(currentConnection, this::checkLeaderAndRedirect); + partitionsClient = new PartitionsTcpClient(currentConnection); }); } @@ -238,11 +254,12 @@ public class AsyncIggyTcpClient { return CompletableFuture.failedFuture( IggyServerException.fromTcpResponse(INVALID_COMMAND_ERROR_CODE, new byte[0])); } - if (connection == null) { + AsyncTcpConnection currentConnection = connection.get(); + if (currentConnection == null) { throw new IggyNotConnectedException(); } - return connection.send(code, Unpooled.copiedBuffer(payload)).thenApply(response -> { + return currentConnection.send(code, Unpooled.copiedBuffer(payload)).thenApply(response -> { try { if (response.readableBytes() <= 1) { return new byte[0]; @@ -382,12 +399,142 @@ public class AsyncIggyTcpClient { * @return a {@link CompletableFuture} that completes when all resources are released */ public CompletableFuture<Void> close() { - if (connection != null) { - return connection.close(); + closed = true; + AsyncTcpConnection currentConnection = connection.get(); + if (currentConnection != null) { + return currentConnection.close(); } return CompletableFuture.completedFuture(null); } + /** + * Returns the server address this client currently targets. Leader + * redirection can change it after login, so it may differ from the + * address the client was built with. + * + * @return the current {@link ConnectionInfo} + */ + public ConnectionInfo getConnectionInfo() { + return connectionInfo; + } + + private AsyncTcpConnection openConnection(ConnectionInfo target) { + return new AsyncTcpConnection( + target.host(), target.port(), enableTls, tlsCertificate, poolConfig, connectionTimeout); + } + + /** + * Post-login leader check, serialized across concurrent logins: fetches + * the cluster roster and, while a healthy leader lives elsewhere, + * reconnects to it and replays the login. A queued login waits for the + * in-flight redirection and then re-checks the roster, so it either + * confirms the new node or retries a redirection that failed. Every + * failure except the replayed login's own is non-fatal; the client then + * stays on the current node. Each check runs with a fresh redirection + * budget, so hitting the cap parks only that login, not the client. + */ + private CompletableFuture<IdentityInfo> checkLeaderAndRedirect(Supplier<CompletableFuture<IdentityInfo>> reLogin) { + CompletableFuture<Void> gate = new CompletableFuture<>(); + CompletableFuture<Void> previous = redirectChain.getAndSet(gate); + LeaderRedirectionState redirectionState = new LeaderRedirectionState(); + return previous.thenCompose(ignored -> redirectToLeader(reLogin, null, redirectionState)) + .whenComplete((identity, error) -> gate.complete(null)); + } + + /** + * One redirection hop: when the roster names a healthy leader elsewhere, + * reconnect to it, replay the login and re-check from the new node, since + * mid-election metadata can point at a node that is itself not the + * leader. Bounded by the per-check redirection budget. + */ + private CompletableFuture<IdentityInfo> redirectToLeader( + Supplier<CompletableFuture<IdentityInfo>> reLogin, + IdentityInfo redirectedIdentity, + LeaderRedirectionState redirectionState) { + ConnectionInfo currentTarget = connectionInfo; + return findLeaderElsewhere(currentTarget).thenCompose(leaderTarget -> { + if (leaderTarget.isEmpty()) { + return CompletableFuture.completedFuture(redirectedIdentity); + } + if (!redirectionState.canRedirect()) { + log.warn( + "Maximum leader redirections ({}) reached, connection will continue on server node {}", + LeaderAwareness.MAX_LEADER_REDIRECTS, + currentTarget.serverAddress()); + return CompletableFuture.completedFuture(redirectedIdentity); + } + return retarget(leaderTarget.get()) + .handle((ignored, error) -> { + if (error != null) { + log.warn( + "Failed to reconnect to leader at {}: {}, connection will continue" + + " on server node {}", + leaderTarget.get().serverAddress(), + error.getMessage(), + currentTarget.serverAddress()); + return CompletableFuture.completedFuture(redirectedIdentity); + } + redirectionState.recordRedirect(); + return reLogin.get() + .thenCompose(identity -> redirectToLeader(reLogin, identity, redirectionState)); + }) + .thenCompose(Function.identity()); + }); + } + + /** + * Picks a healthy leader living elsewhere from the roster, waiting out a + * transiently leaderless election. Resolves to empty on any failure + * (metadata fetch, malformed roster) so the redirection path never fails + * the login that triggered it. + */ + private CompletableFuture<Optional<ConnectionInfo>> findLeaderElsewhere(ConnectionInfo currentTarget) { + SystemClient currentSystemClient = systemClient; + if (currentSystemClient == null) { + return CompletableFuture.completedFuture(Optional.empty()); + } + return LeaderAwareness.findLeaderElsewhere(currentSystemClient::getClusterMetadata, currentTarget); + } + + private CompletableFuture<Void> retarget(ConnectionInfo newTarget) { + AsyncTcpConnection oldConnection = connection.get(); + AsyncTcpConnection newConnection; + try { + newConnection = openConnection(newTarget); + } catch (RuntimeException error) { + return CompletableFuture.failedFuture(error); + } + AsyncTcpConnection openedConnection = newConnection; + return newConnection + .connect() + .thenCompose(ignored -> publishConnection(oldConnection, openedConnection, newTarget)) + .whenComplete((ignored, error) -> { + if (error != null) { + openedConnection.close(); + } + }); + } + + private CompletableFuture<Void> publishConnection( + AsyncTcpConnection oldConnection, AsyncTcpConnection newConnection, ConnectionInfo newTarget) { + if (!connection.compareAndSet(oldConnection, newConnection)) { + return CompletableFuture.failedFuture( + new IllegalStateException("Connection replaced concurrently during leader redirection")); + } + if (closed) { + oldConnection.close(); + return CompletableFuture.failedFuture( + new IggyNotConnectedException("Client closed during leader redirection")); + } + connectionInfo = newTarget; + oldConnection.close().whenComplete((ignored, closeError) -> { + if (closeError != null) { + log.warn("Failed to close previous connection: {}", closeError.getMessage()); + } + }); + return CompletableFuture.completedFuture(null); + } + private static boolean isSessionControlCode(int code) { return code == CommandCode.User.LOGIN.getValue() || code == CommandCode.User.LOGOUT.getValue() diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java index c74c615d4..38eb30f05 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java @@ -84,10 +84,8 @@ public class AsyncTcpConnection { int port, boolean enableTls, Optional<File> tlsCertificate, - TCPConnectionPoolConfig poolConfig, + TcpConnectionPoolConfig poolConfig, Optional<Duration> connectionTimeout) { - this.eventLoopGroup = new MultiThreadIoEventLoopGroup(NioIoHandler.newFactory()); - SslContext sslContext = null; if (enableTls) { try { @@ -99,6 +97,8 @@ public class AsyncTcpConnection { } } + this.eventLoopGroup = new MultiThreadIoEventLoopGroup(NioIoHandler.newFactory()); + var bootstrap = new Bootstrap() .group(eventLoopGroup) .channel(NioSocketChannel.class) @@ -195,11 +195,12 @@ public class AsyncTcpConnection { public CompletableFuture<ByteBuf> send(int commandCode, ByteBuf payload) { captureLoginPayloadIfNeeded(commandCode, payload); CompletableFuture<ByteBuf> responseFuture = new CompletableFuture<>(); + CompletableFuture<ByteBuf> callerFuture = new CompletableFuture<>(); channelPool.acquire().addListener((FutureListener<Channel>) f -> { if (!f.isSuccess()) { payload.release(); - responseFuture.completeExceptionally(mapAcquireException(f.cause())); + callerFuture.completeExceptionally(mapAcquireException(f.cause())); return; } @@ -210,9 +211,17 @@ public class AsyncTcpConnection { && commandCode != CommandCode.System.PING.getValue() && commandCode != CommandCode.System.GET_STATS.getValue(); - responseFuture.handle((res, ex) -> { - handlePostResponse(channel, commandCode, isLoginCommand, ex); - return null; + responseFuture.whenComplete((response, error) -> { + try { + handlePostResponse(channel, commandCode, isLoginCommand, error); + } catch (RuntimeException bookkeepingError) { + log.error("Post-response bookkeeping failed: {}", bookkeepingError.getMessage()); + } + if (error != null) { + callerFuture.completeExceptionally(error); + } else { + callerFuture.complete(response); + } }); CompletableFuture<Void> authStep; @@ -242,7 +251,7 @@ public class AsyncTcpConnection { }); }); - return responseFuture; + return callerFuture; } private static Throwable mapAcquireException(Throwable cause) { @@ -396,34 +405,48 @@ public class AsyncTcpConnection { } } + @Override + public void channelInactive(ChannelHandlerContext ctx) { + failPendingRequests(new IggyConnectionException("Connection closed before a response arrived")); + ctx.fireChannelInactive(); + } + @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { - CompletableFuture<ByteBuf> f; - while ((f = responseQueue.poll()) != null) { - f.completeExceptionally(cause); - } + failPendingRequests(cause); ctx.close(); } + + private void failPendingRequests(Throwable cause) { + CompletableFuture<ByteBuf> pending; + while ((pending = responseQueue.poll()) != null) { + pending.completeExceptionally(cause); + } + } } - public static class TCPConnectionPoolConfig { + public static class TcpConnectionPoolConfig { private final int maxConnections; private final int maxPendingAcquires; private final long acquireTimeoutMillis; - public TCPConnectionPoolConfig() { + public TcpConnectionPoolConfig() { this( - Builder.DEFAULT_MAX_CONNECTION, - Builder.DEFAULT_MAX_PENDING_ACQUIRES, - Builder.DEFAULT_ACQUIRE_TIMEOUT_MILLIS); + TcpConnectionPoolConfigBuilder.DEFAULT_MAX_CONNECTION, + TcpConnectionPoolConfigBuilder.DEFAULT_MAX_PENDING_ACQUIRES, + TcpConnectionPoolConfigBuilder.DEFAULT_ACQUIRE_TIMEOUT_MILLIS); } - public TCPConnectionPoolConfig(int maxConnections, int maxPendingAcquires, long acquireTimeoutMillis) { + public TcpConnectionPoolConfig(int maxConnections, int maxPendingAcquires, long acquireTimeoutMillis) { this.maxConnections = maxConnections; this.maxPendingAcquires = maxPendingAcquires; this.acquireTimeoutMillis = acquireTimeoutMillis; } + public static TcpConnectionPoolConfigBuilder builder() { + return new TcpConnectionPoolConfigBuilder(); + } + public int getMaxConnections() { return this.maxConnections; } @@ -436,7 +459,7 @@ public class AsyncTcpConnection { return this.acquireTimeoutMillis; } - public static final class Builder { + public static final class TcpConnectionPoolConfigBuilder { public static final int DEFAULT_MAX_CONNECTION = 5; public static final int DEFAULT_MAX_PENDING_ACQUIRES = 1000; public static final int DEFAULT_ACQUIRE_TIMEOUT_MILLIS = 3000; @@ -445,9 +468,9 @@ public class AsyncTcpConnection { private int maxPendingAcquires; private long acquireTimeoutMillis; - public Builder() {} + public TcpConnectionPoolConfigBuilder() {} - public Builder setMaxConnections(int maxConnections) { + public TcpConnectionPoolConfigBuilder setMaxConnections(int maxConnections) { if (maxConnections <= 0) { throw new IggyInvalidArgumentException("Connection pool size cannot be 0 or negative"); } @@ -455,7 +478,7 @@ public class AsyncTcpConnection { return this; } - public Builder setMaxPendingAcquires(int maxPendingAcquires) { + public TcpConnectionPoolConfigBuilder setMaxPendingAcquires(int maxPendingAcquires) { if (maxPendingAcquires <= 0) { throw new IggyInvalidArgumentException("Max Pending Acquires cannot be 0 or negative"); } @@ -463,7 +486,7 @@ public class AsyncTcpConnection { return this; } - public Builder setAcquireTimeoutMillis(long acquireTimeoutMillis) { + public TcpConnectionPoolConfigBuilder setAcquireTimeoutMillis(long acquireTimeoutMillis) { if (acquireTimeoutMillis <= 0) { throw new IggyInvalidArgumentException("Acquire timeout cannot be 0 or negative"); } @@ -471,7 +494,7 @@ public class AsyncTcpConnection { return this; } - public TCPConnectionPoolConfig build() { + public TcpConnectionPoolConfig build() { if (this.maxConnections == 0) { this.maxConnections = DEFAULT_MAX_CONNECTION; } @@ -481,7 +504,7 @@ public class AsyncTcpConnection { if (this.maxPendingAcquires == 0) { this.maxPendingAcquires = DEFAULT_MAX_PENDING_ACQUIRES; } - return new TCPConnectionPoolConfig(maxConnections, maxPendingAcquires, acquireTimeoutMillis); + return new TcpConnectionPoolConfig(maxConnections, maxPendingAcquires, acquireTimeoutMillis); } } } diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/ConsumerGroupsTcpClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/ConsumerGroupsTcpClient.java index 71133f155..9b4561fc6 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/ConsumerGroupsTcpClient.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/ConsumerGroupsTcpClient.java @@ -36,6 +36,7 @@ import java.util.ArrayList; import java.util.List; import java.util.Optional; import java.util.concurrent.CompletableFuture; +import java.util.function.Supplier; /** * Async TCP implementation of consumer groups client. @@ -43,10 +44,14 @@ import java.util.concurrent.CompletableFuture; public class ConsumerGroupsTcpClient implements ConsumerGroupsClient { private static final Logger log = LoggerFactory.getLogger(ConsumerGroupsTcpClient.class); - private final AsyncTcpConnection connection; + private final Supplier<AsyncTcpConnection> connectionSupplier; - public ConsumerGroupsTcpClient(AsyncTcpConnection connection) { - this.connection = connection; + public ConsumerGroupsTcpClient(Supplier<AsyncTcpConnection> connectionSupplier) { + this.connectionSupplier = connectionSupplier; + } + + private AsyncTcpConnection connection() { + return connectionSupplier.get(); } @Override @@ -58,7 +63,7 @@ public class ConsumerGroupsTcpClient implements ConsumerGroupsClient { log.debug("Getting consumer group - Stream: {}, Topic: {}, Group: {}", streamId, topicId, groupId); - return connection + return connection() .send(CommandCode.ConsumerGroup.GET.getValue(), payload) .thenApply(response -> { try { @@ -80,7 +85,7 @@ public class ConsumerGroupsTcpClient implements ConsumerGroupsClient { log.debug("Getting consumer groups - Stream: {}, Topic: {}", streamId, topicId); - return connection + return connection() .send(CommandCode.ConsumerGroup.GET_ALL.getValue(), payload) .thenApply(response -> { try { @@ -108,7 +113,7 @@ public class ConsumerGroupsTcpClient implements ConsumerGroupsClient { log.debug("Creating consumer group - Stream: {}, Topic: {}, Name: {}", streamId, topicId, name); - return connection + return connection() .send(CommandCode.ConsumerGroup.CREATE.getValue(), payload) .thenApply(response -> { try { @@ -127,7 +132,7 @@ public class ConsumerGroupsTcpClient implements ConsumerGroupsClient { log.debug("Deleting consumer group - Stream: {}, Topic: {}, Group: {}", streamId, topicId, groupId); - return connection + return connection() .send(CommandCode.ConsumerGroup.DELETE.getValue(), payload) .thenAccept(response -> { response.release(); @@ -149,7 +154,7 @@ public class ConsumerGroupsTcpClient implements ConsumerGroupsClient { log.debug("Joining consumer group - Stream: {}, Topic: {}, Group: {}", streamId, topicId, groupId); - return connection + return connection() .send(CommandCode.ConsumerGroup.JOIN.getValue(), payload) .thenAccept(response -> { log.debug("Successfully joined consumer group"); @@ -172,7 +177,7 @@ public class ConsumerGroupsTcpClient implements ConsumerGroupsClient { log.debug("Leaving consumer group - Stream: {}, Topic: {}, Group: {}", streamId, topicId, groupId); - return connection + return connection() .send(CommandCode.ConsumerGroup.LEAVE.getValue(), payload) .thenAccept(response -> { log.debug("Successfully left consumer group"); diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/ConsumerOffsetsTcpClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/ConsumerOffsetsTcpClient.java index aa6c6e41c..cb53a72e2 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/ConsumerOffsetsTcpClient.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/ConsumerOffsetsTcpClient.java @@ -33,6 +33,7 @@ import org.slf4j.LoggerFactory; import java.math.BigInteger; import java.util.Optional; import java.util.concurrent.CompletableFuture; +import java.util.function.Supplier; /** * Async TCP implementation of consumer offsets client. @@ -40,10 +41,14 @@ import java.util.concurrent.CompletableFuture; public class ConsumerOffsetsTcpClient implements ConsumerOffsetsClient { private static final Logger log = LoggerFactory.getLogger(ConsumerOffsetsTcpClient.class); - private final AsyncTcpConnection connection; + private final Supplier<AsyncTcpConnection> connectionSupplier; - public ConsumerOffsetsTcpClient(AsyncTcpConnection connection) { - this.connection = connection; + public ConsumerOffsetsTcpClient(Supplier<AsyncTcpConnection> connectionSupplier) { + this.connectionSupplier = connectionSupplier; + } + + private AsyncTcpConnection connection() { + return connectionSupplier.get(); } @Override @@ -63,7 +68,7 @@ public class ConsumerOffsetsTcpClient implements ConsumerOffsetsClient { consumer, offset); - return connection + return connection() .send(CommandCode.ConsumerOffset.STORE.getValue(), payload) .thenAccept(response -> { response.release(); @@ -85,7 +90,7 @@ public class ConsumerOffsetsTcpClient implements ConsumerOffsetsClient { partitionId, consumer); - return connection + return connection() .send(CommandCode.ConsumerOffset.GET.getValue(), payload) .thenApply(response -> { try { diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/LeaderAwareness.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/LeaderAwareness.java new file mode 100644 index 000000000..f541e9de7 --- /dev/null +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/LeaderAwareness.java @@ -0,0 +1,282 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iggy.client.async.tcp; + +import org.apache.iggy.client.ConnectionInfo; +import org.apache.iggy.cluster.ClusterMetadata; +import org.apache.iggy.cluster.ClusterNode; +import org.apache.iggy.cluster.ClusterNodeRole; +import org.apache.iggy.cluster.ClusterNodeStatus; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.net.InetAddress; +import java.net.UnknownHostException; +import java.time.Duration; +import java.util.Arrays; +import java.util.List; +import java.util.Locale; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Executor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Function; +import java.util.function.Supplier; + +/** + * Leader selection logic for cluster-aware TCP clients. + */ +final class LeaderAwareness { + + static final int MAX_LEADER_REDIRECTS = 3; + + /** + * How long to wait for a transiently leaderless cluster to elect before + * proceeding on the current node anyway. A restarted node cedes the + * primaryship its stale view assigns it, and until the peers' election + * completes the roster reports no leader; that window is roughly one + * heartbeat timeout. + */ + static final Duration LEADERLESS_WAIT_BUDGET = Duration.ofSeconds(5); + + static final Duration LEADERLESS_POLL_INTERVAL = Duration.ofMillis(250); + + private static final Logger log = LoggerFactory.getLogger(LeaderAwareness.class); + + private LeaderAwareness() {} + + /** + * Finds a healthy leader living elsewhere, polling the roster through a + * transiently leaderless cluster for up to {@link #LEADERLESS_WAIT_BUDGET}. + * Resolves to empty when the client should stay on the current node: it + * already is the leader, the cluster is single-node, no leader appeared + * within the budget, or the roster could not be fetched. Never resolves + * exceptionally, so the redirection path cannot fail the login that + * triggered it. + */ + static CompletableFuture<Optional<ConnectionInfo>> findLeaderElsewhere( + Supplier<CompletableFuture<ClusterMetadata>> fetchMetadata, ConnectionInfo currentTarget) { + return findLeaderElsewhere(fetchMetadata, currentTarget, LEADERLESS_WAIT_BUDGET, LEADERLESS_POLL_INTERVAL); + } + + static CompletableFuture<Optional<ConnectionInfo>> findLeaderElsewhere( + Supplier<CompletableFuture<ClusterMetadata>> fetchMetadata, + ConnectionInfo currentTarget, + Duration leaderlessWaitBudget, + Duration leaderlessPollInterval) { + long electionDeadlineNanos = System.nanoTime() + leaderlessWaitBudget.toNanos(); + return pollForLeader( + fetchMetadata, currentTarget, leaderlessWaitBudget, leaderlessPollInterval, electionDeadlineNanos); + } + + private static CompletableFuture<Optional<ConnectionInfo>> pollForLeader( + Supplier<CompletableFuture<ClusterMetadata>> fetchMetadata, + ConnectionInfo currentTarget, + Duration leaderlessWaitBudget, + Duration leaderlessPollInterval, + long electionDeadlineNanos) { + CompletableFuture<ClusterMetadata> fetched; + try { + fetched = fetchMetadata.get(); + } catch (RuntimeException fetchError) { + fetched = CompletableFuture.failedFuture(fetchError); + } + return fetched.<CompletableFuture<Optional<ConnectionInfo>>>handleAsync((metadata, error) -> { + if (error != null) { + log.warn( + "Failed to get cluster metadata: {}, connection will continue on server node {}", + error.getMessage(), + currentTarget.serverAddress()); + return CompletableFuture.completedFuture(Optional.empty()); + } + LeaderCheck check; + try { + check = checkLeader(metadata, currentTarget); + } catch (RuntimeException selectionError) { + log.warn( + "Failed to select leader from cluster metadata: {}, connection will continue" + + " on server node {}", + selectionError.getMessage(), + currentTarget.serverAddress()); + return CompletableFuture.completedFuture(Optional.empty()); + } + if (check instanceof LeaderCheck.Redirect redirect) { + return CompletableFuture.completedFuture(Optional.of(redirect.target())); + } + if (check instanceof LeaderCheck.NoLeader) { + if (System.nanoTime() >= electionDeadlineNanos) { + log.warn( + "No active leader found in cluster metadata within {}, connection will" + + " continue on server node {}", + leaderlessWaitBudget, + currentTarget.serverAddress()); + return CompletableFuture.completedFuture(Optional.empty()); + } + Executor retryAfterInterval = CompletableFuture.delayedExecutor( + leaderlessPollInterval.toMillis(), TimeUnit.MILLISECONDS); + return CompletableFuture.supplyAsync( + () -> pollForLeader( + fetchMetadata, + currentTarget, + leaderlessWaitBudget, + leaderlessPollInterval, + electionDeadlineNanos), + retryAfterInterval) + .thenCompose(Function.identity()); + } + return CompletableFuture.completedFuture(Optional.empty()); + }) + .thenCompose(Function.identity()); + } + + /** + * One leader-check verdict from a cluster-metadata snapshot. + */ + static LeaderCheck checkLeader(ClusterMetadata metadata, ConnectionInfo currentTarget) { + var nodes = metadata.nodes(); + if (nodes.size() == 1) { + log.debug( + "Single-node cluster detected ({}), no leader redirection needed", + nodes.get(0).name()); + return new LeaderCheck.Stay(); + } + + ClusterNode leader = nodes.stream() + .filter(node -> node.role() == ClusterNodeRole.Leader && node.status() == ClusterNodeStatus.Healthy) + .findFirst() + .orElse(null); + if (leader == null) { + log.debug( + "No active leader found in cluster metadata for current target {}", currentTarget.serverAddress()); + return new LeaderCheck.NoLeader(); + } + + var leaderTcpPort = leader.endpoints().tcp(); + if (leaderTcpPort == 0) { + log.warn( + "Leader node {} has the tcp transport disabled, connection will continue on server node {}", + leader.name(), + currentTarget.serverAddress()); + return new LeaderCheck.Stay(); + } + + var leaderTarget = new ConnectionInfo(leader.ip(), leaderTcpPort); + if (isSameAddress(currentTarget, leaderTarget)) { + log.debug("Already connected to leader at {}", currentTarget.serverAddress()); + return new LeaderCheck.Stay(); + } + + log.info( + "Found leader node {} at {}, current connection to {} is not the leader", + leader.name(), + leaderTarget.serverAddress(), + currentTarget.serverAddress()); + return new LeaderCheck.Redirect(leaderTarget); + } + + /** + * Whether two targets name the same endpoint. Ports must match; hosts + * match when they are textually equal after canonicalization, when both + * only reach the local machine, or when they resolve to a common + * address. A hostname the resolver does not know compares unequal, which + * at worst costs one redirect hop back to the same node. + */ + static boolean isSameAddress(ConnectionInfo target1, ConnectionInfo target2) { + if (target1.port() != target2.port()) { + return false; + } + var host1 = canonicalHost(target1.host()); + var host2 = canonicalHost(target2.host()); + if (host1.equals(host2)) { + return true; + } + return resolveToSameHost(host1, host2); + } + + private static String canonicalHost(String host) { + var canonical = host.toLowerCase(Locale.ROOT); + if (canonical.startsWith("[") && canonical.endsWith("]")) { + canonical = canonical.substring(1, canonical.length() - 1); + } + return canonical; + } + + private static boolean resolveToSameHost(String host1, String host2) { + InetAddress[] addresses1; + InetAddress[] addresses2; + try { + addresses1 = InetAddress.getAllByName(host1); + addresses2 = InetAddress.getAllByName(host2); + } catch (UnknownHostException resolutionError) { + log.debug("Could not compare hosts {} and {} by address: {}", host1, host2, resolutionError.getMessage()); + return false; + } + if (reachesOnlyLocalMachine(addresses1) && reachesOnlyLocalMachine(addresses2)) { + return true; + } + List<InetAddress> candidates = Arrays.asList(addresses2); + return Arrays.stream(addresses1).anyMatch(candidates::contains); + } + + /** + * Loopback and unspecified addresses all reach the local machine, so + * localhost, 127.0.0.1, ::1 and [::] name the same endpoint on a given + * port. + */ + private static boolean reachesOnlyLocalMachine(InetAddress[] addresses) { + return Arrays.stream(addresses).allMatch(address -> address.isLoopbackAddress() || address.isAnyLocalAddress()); + } + + /** + * One leader-check verdict from a cluster-metadata snapshot. + */ + sealed interface LeaderCheck { + + /** A healthy leader with an enabled tcp transport lives elsewhere; reconnect to it. */ + record Redirect(ConnectionInfo target) implements LeaderCheck {} + + /** Stay on the current node: it is the leader, the cluster is single-node, or the leader is unusable. */ + record Stay() implements LeaderCheck {} + + /** No healthy leader is marked, e.g. mid-election; worth re-polling. */ + record NoLeader() implements LeaderCheck {} + } + + /** + * Counts the redirection hops of a single leader check so two nodes that + * both claim leadership cannot ping-pong a client forever. Failed + * reconnect attempts do not count. The state lives for one check only; + * the next login starts with a fresh budget, so exhausting the cap never + * parks the client on a follower for its whole lifetime. + */ + static final class LeaderRedirectionState { + + private final AtomicInteger redirectCount = new AtomicInteger(); + + boolean canRedirect() { + return redirectCount.get() < MAX_LEADER_REDIRECTS; + } + + void recordRedirect() { + redirectCount.incrementAndGet(); + } + } +} diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/LoginRedirectionHook.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/LoginRedirectionHook.java new file mode 100644 index 000000000..d1c269f9e --- /dev/null +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/LoginRedirectionHook.java @@ -0,0 +1,47 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iggy.client.async.tcp; + +import org.apache.iggy.user.IdentityInfo; + +import java.util.concurrent.CompletableFuture; +import java.util.function.Supplier; + +/** + * Runs after a successful login to redirect the client to the cluster + * leader when it is connected to a follower. + */ +@FunctionalInterface +interface LoginRedirectionHook { + + LoginRedirectionHook NONE = reLogin -> CompletableFuture.completedFuture(null); + + /** + * Checks the cluster roster and, while the current node is not the + * leader, retargets the connection and re-runs the login. + * + * @param reLogin replays the just-completed login against the current + * target; it must not trigger another redirection check itself, since + * the hook drives any further hops and serializes concurrent checks + * @return the redirected login's identity, or {@code null} when the client + * stays on the current node + */ + CompletableFuture<IdentityInfo> afterLogin(Supplier<CompletableFuture<IdentityInfo>> reLogin); +} diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/MessagesTcpClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/MessagesTcpClient.java index 51c84f375..5cb2d12fc 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/MessagesTcpClient.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/MessagesTcpClient.java @@ -34,6 +34,7 @@ import org.apache.iggy.serde.CommandCode; import java.util.List; import java.util.Optional; import java.util.concurrent.CompletableFuture; +import java.util.function.Supplier; import static org.apache.iggy.serde.BytesSerializer.toBytes; @@ -42,10 +43,14 @@ import static org.apache.iggy.serde.BytesSerializer.toBytes; */ public class MessagesTcpClient implements MessagesClient { - private final AsyncTcpConnection connection; + private final Supplier<AsyncTcpConnection> connectionSupplier; - public MessagesTcpClient(AsyncTcpConnection connection) { - this.connection = connection; + public MessagesTcpClient(Supplier<AsyncTcpConnection> connectionSupplier) { + this.connectionSupplier = connectionSupplier; + } + + private AsyncTcpConnection connection() { + return connectionSupplier.get(); } @Override @@ -79,7 +84,7 @@ public class MessagesTcpClient implements MessagesClient { payload.writeByte(autoCommit ? 1 : 0); // Send async request and transform response - return connection.send(CommandCode.Messages.POLL.getValue(), payload).thenApply(response -> { + return connection().send(CommandCode.Messages.POLL.getValue(), payload).thenApply(response -> { try { return BytesDeserializer.readPolledMessages(response); } finally { @@ -123,7 +128,7 @@ public class MessagesTcpClient implements MessagesClient { } // Send async request (no response data expected for send) - return connection.send(CommandCode.Messages.SEND.getValue(), payload).thenAccept(response -> { + return connection().send(CommandCode.Messages.SEND.getValue(), payload).thenAccept(response -> { // Response received, messages sent successfully response.release(); // Release the buffer }); diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/PartitionsTcpClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/PartitionsTcpClient.java index c142bf683..e8544067e 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/PartitionsTcpClient.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/PartitionsTcpClient.java @@ -28,6 +28,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.concurrent.CompletableFuture; +import java.util.function.Supplier; /** * Async TCP implementation of partitions client. @@ -35,10 +36,14 @@ import java.util.concurrent.CompletableFuture; public class PartitionsTcpClient implements PartitionsClient { private static final Logger log = LoggerFactory.getLogger(PartitionsTcpClient.class); - private final AsyncTcpConnection connection; + private final Supplier<AsyncTcpConnection> connectionSupplier; - public PartitionsTcpClient(AsyncTcpConnection connection) { - this.connection = connection; + public PartitionsTcpClient(Supplier<AsyncTcpConnection> connectionSupplier) { + this.connectionSupplier = connectionSupplier; + } + + private AsyncTcpConnection connection() { + return connectionSupplier.get(); } @Override @@ -49,9 +54,11 @@ public class PartitionsTcpClient implements PartitionsClient { log.debug("Creating {} partitions for stream: {}, topic: {}", partitionsCount, streamId, topicId); - return connection.send(CommandCode.Partition.CREATE.getValue(), payload).thenAccept(response -> { - response.release(); - }); + return connection() + .send(CommandCode.Partition.CREATE.getValue(), payload) + .thenAccept(response -> { + response.release(); + }); } @Override @@ -62,8 +69,10 @@ public class PartitionsTcpClient implements PartitionsClient { log.debug("Deleting {} partitions for stream: {}, topic: {}", partitionsCount, streamId, topicId); - return connection.send(CommandCode.Partition.DELETE.getValue(), payload).thenAccept(response -> { - response.release(); - }); + return connection() + .send(CommandCode.Partition.DELETE.getValue(), payload) + .thenAccept(response -> { + response.release(); + }); } } diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/PersonalAccessTokensTcpClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/PersonalAccessTokensTcpClient.java index fe31a549d..b5b97842f 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/PersonalAccessTokensTcpClient.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/PersonalAccessTokensTcpClient.java @@ -35,6 +35,7 @@ import java.util.ArrayList; import java.util.List; import java.util.Optional; import java.util.concurrent.CompletableFuture; +import java.util.function.Supplier; /** * Async TCP implementation of personal access tokens client. @@ -42,10 +43,21 @@ import java.util.concurrent.CompletableFuture; public class PersonalAccessTokensTcpClient implements PersonalAccessTokensClient { private static final Logger log = LoggerFactory.getLogger(PersonalAccessTokensTcpClient.class); - private final AsyncTcpConnection connection; + private final Supplier<AsyncTcpConnection> connectionSupplier; + private final LoginRedirectionHook redirectionHook; - public PersonalAccessTokensTcpClient(AsyncTcpConnection connection) { - this.connection = connection; + public PersonalAccessTokensTcpClient(Supplier<AsyncTcpConnection> connectionSupplier) { + this(connectionSupplier, LoginRedirectionHook.NONE); + } + + PersonalAccessTokensTcpClient( + Supplier<AsyncTcpConnection> connectionSupplier, LoginRedirectionHook redirectionHook) { + this.connectionSupplier = connectionSupplier; + this.redirectionHook = redirectionHook; + } + + private AsyncTcpConnection connection() { + return connectionSupplier.get(); } @Override @@ -56,7 +68,7 @@ public class PersonalAccessTokensTcpClient implements PersonalAccessTokensClient log.debug("Creating personal access token: {}", name); - return connection + return connection() .send(CommandCode.PersonalAccessToken.CREATE.getValue(), payload) .thenApply(response -> { try { @@ -73,7 +85,7 @@ public class PersonalAccessTokensTcpClient implements PersonalAccessTokensClient log.debug("Getting all personal access tokens"); - return connection + return connection() .send(CommandCode.PersonalAccessToken.GET_ALL.getValue(), payload) .thenApply(response -> { try { @@ -94,7 +106,7 @@ public class PersonalAccessTokensTcpClient implements PersonalAccessTokensClient log.debug("Deleting personal access token: {}", name); - return connection + return connection() .send(CommandCode.PersonalAccessToken.DELETE.getValue(), payload) .thenAccept(response -> { response.release(); @@ -103,11 +115,17 @@ public class PersonalAccessTokensTcpClient implements PersonalAccessTokensClient @Override public CompletableFuture<IdentityInfo> loginWithPersonalAccessToken(String token) { + return loginWithoutRedirect(token).thenCompose(identity -> redirectionHook + .afterLogin(() -> loginWithoutRedirect(token)) + .thenApply(redirectedIdentity -> redirectedIdentity != null ? redirectedIdentity : identity)); + } + + private CompletableFuture<IdentityInfo> loginWithoutRedirect(String token) { var payload = BytesSerializer.toBytes(token); log.debug("Logging in with personal access token"); - return connection + return connection() .send(CommandCode.PersonalAccessToken.LOGIN.getValue(), payload) .thenApply(response -> { try { diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/StreamsTcpClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/StreamsTcpClient.java index 0a0466881..91f37e4a5 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/StreamsTcpClient.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/StreamsTcpClient.java @@ -32,6 +32,7 @@ import java.util.ArrayList; import java.util.List; import java.util.Optional; import java.util.concurrent.CompletableFuture; +import java.util.function.Supplier; import static org.apache.iggy.serde.BytesDeserializer.readStreamBase; import static org.apache.iggy.serde.BytesDeserializer.readStreamDetails; @@ -42,10 +43,14 @@ import static org.apache.iggy.serde.BytesSerializer.toBytes; */ public class StreamsTcpClient implements StreamsClient { - private final AsyncTcpConnection connection; + private final Supplier<AsyncTcpConnection> connectionSupplier; - public StreamsTcpClient(AsyncTcpConnection connection) { - this.connection = connection; + public StreamsTcpClient(Supplier<AsyncTcpConnection> connectionSupplier) { + this.connectionSupplier = connectionSupplier; + } + + private AsyncTcpConnection connection() { + return connectionSupplier.get(); } @Override @@ -55,7 +60,7 @@ public class StreamsTcpClient implements StreamsClient { payload.writeBytes(BytesSerializer.toBytes(name)); - return connection.send(CommandCode.Stream.CREATE.getValue(), payload).thenApply(response -> { + return connection().send(CommandCode.Stream.CREATE.getValue(), payload).thenApply(response -> { StreamDetails details = readStreamDetails(response); response.release(); return details; @@ -66,7 +71,7 @@ public class StreamsTcpClient implements StreamsClient { public CompletableFuture<Optional<StreamDetails>> getStream(StreamId streamId) { var payload = toBytes(streamId); - return connection.send(CommandCode.Stream.GET.getValue(), payload).thenApply(response -> { + return connection().send(CommandCode.Stream.GET.getValue(), payload).thenApply(response -> { Optional<StreamDetails> result; if (response.isReadable()) { result = Optional.of(readStreamDetails(response)); @@ -80,7 +85,7 @@ public class StreamsTcpClient implements StreamsClient { @Override public CompletableFuture<List<StreamBase>> getStreams() { - return connection + return connection() .send(CommandCode.Stream.GET_ALL.getValue(), Unpooled.EMPTY_BUFFER) .thenApply(response -> { List<StreamBase> streams = new ArrayList<>(); @@ -101,13 +106,13 @@ public class StreamsTcpClient implements StreamsClient { payload.writeBytes(idBytes); payload.writeBytes(BytesSerializer.toBytes(name)); - return connection.send(CommandCode.Stream.UPDATE.getValue(), payload).thenAccept(ReferenceCounted::release); + return connection().send(CommandCode.Stream.UPDATE.getValue(), payload).thenAccept(ReferenceCounted::release); } @Override public CompletableFuture<Void> deleteStream(StreamId streamId) { var payload = toBytes(streamId); - return connection.send(CommandCode.Stream.DELETE.getValue(), payload).thenAccept(ReferenceCounted::release); + return connection().send(CommandCode.Stream.DELETE.getValue(), payload).thenAccept(ReferenceCounted::release); } } diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/SystemTcpClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/SystemTcpClient.java index acf0db698..8692a2065 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/SystemTcpClient.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/SystemTcpClient.java @@ -21,6 +21,7 @@ package org.apache.iggy.client.async.tcp; import io.netty.buffer.Unpooled; import org.apache.iggy.client.async.SystemClient; +import org.apache.iggy.cluster.ClusterMetadata; import org.apache.iggy.serde.BytesDeserializer; import org.apache.iggy.serde.CommandCode; import org.apache.iggy.system.ClientInfo; @@ -32,6 +33,7 @@ import org.slf4j.LoggerFactory; import java.util.ArrayList; import java.util.List; import java.util.concurrent.CompletableFuture; +import java.util.function.Supplier; /** * Async TCP implementation of system client. @@ -39,10 +41,14 @@ import java.util.concurrent.CompletableFuture; public class SystemTcpClient implements SystemClient { private static final Logger log = LoggerFactory.getLogger(SystemTcpClient.class); - private final AsyncTcpConnection connection; + private final Supplier<AsyncTcpConnection> connectionSupplier; - public SystemTcpClient(AsyncTcpConnection connection) { - this.connection = connection; + public SystemTcpClient(Supplier<AsyncTcpConnection> connectionSupplier) { + this.connectionSupplier = connectionSupplier; + } + + private AsyncTcpConnection connection() { + return connectionSupplier.get(); } @Override @@ -51,13 +57,32 @@ public class SystemTcpClient implements SystemClient { log.debug("Getting server statistics"); - return connection.send(CommandCode.System.GET_STATS.getValue(), payload).thenApply(response -> { - try { - return BytesDeserializer.readStats(response); - } finally { - response.release(); - } - }); + return connection() + .send(CommandCode.System.GET_STATS.getValue(), payload) + .thenApply(response -> { + try { + return BytesDeserializer.readStats(response); + } finally { + response.release(); + } + }); + } + + @Override + public CompletableFuture<ClusterMetadata> getClusterMetadata() { + var payload = Unpooled.EMPTY_BUFFER; + + log.debug("Getting cluster metadata"); + + return connection() + .send(CommandCode.System.GET_CLUSTER_METADATA.getValue(), payload) + .thenApply(response -> { + try { + return BytesDeserializer.readClusterMetadata(response); + } finally { + response.release(); + } + }); } @Override @@ -66,7 +91,7 @@ public class SystemTcpClient implements SystemClient { log.debug("Getting current client info"); - return connection.send(CommandCode.System.GET_ME.getValue(), payload).thenApply(response -> { + return connection().send(CommandCode.System.GET_ME.getValue(), payload).thenApply(response -> { try { return BytesDeserializer.readClientInfoDetails(response); } finally { @@ -82,7 +107,7 @@ public class SystemTcpClient implements SystemClient { log.debug("Getting client info for client ID: {}", clientId); - return connection + return connection() .send(CommandCode.System.GET_CLIENT.getValue(), payload) .thenApply(response -> { try { @@ -99,7 +124,7 @@ public class SystemTcpClient implements SystemClient { log.debug("Getting all clients"); - return connection + return connection() .send(CommandCode.System.GET_ALL_CLIENTS.getValue(), payload) .thenApply(response -> { try { @@ -120,7 +145,7 @@ public class SystemTcpClient implements SystemClient { log.debug("Pinging server"); - return connection.send(CommandCode.System.PING.getValue(), payload).thenApply(response -> { + return connection().send(CommandCode.System.PING.getValue(), payload).thenApply(response -> { response.release(); return "pong"; }); diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/TopicsTcpClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/TopicsTcpClient.java index dce5b70f3..737f0a341 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/TopicsTcpClient.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/TopicsTcpClient.java @@ -35,6 +35,7 @@ import java.util.ArrayList; import java.util.List; import java.util.Optional; import java.util.concurrent.CompletableFuture; +import java.util.function.Supplier; import static org.apache.iggy.serde.BytesSerializer.toBytes; import static org.apache.iggy.serde.BytesSerializer.toBytesAsU64; @@ -44,10 +45,14 @@ import static org.apache.iggy.serde.BytesSerializer.toBytesAsU64; */ public class TopicsTcpClient implements TopicsClient { - private final AsyncTcpConnection connection; + private final Supplier<AsyncTcpConnection> connectionSupplier; - public TopicsTcpClient(AsyncTcpConnection connection) { - this.connection = connection; + public TopicsTcpClient(Supplier<AsyncTcpConnection> connectionSupplier) { + this.connectionSupplier = connectionSupplier; + } + + private AsyncTcpConnection connection() { + return connectionSupplier.get(); } @Override @@ -56,7 +61,7 @@ public class TopicsTcpClient implements TopicsClient { payload.writeBytes(toBytes(streamId)); payload.writeBytes(toBytes(topicId)); - return connection.send(CommandCode.Topic.GET.getValue(), payload).thenApply(response -> { + return connection().send(CommandCode.Topic.GET.getValue(), payload).thenApply(response -> { try { if (response.isReadable()) { return Optional.of(BytesDeserializer.readTopicDetails(response)); @@ -72,7 +77,7 @@ public class TopicsTcpClient implements TopicsClient { public CompletableFuture<List<Topic>> getTopics(StreamId streamId) { var payload = toBytes(streamId); - return connection.send(CommandCode.Topic.GET_ALL.getValue(), payload).thenApply(response -> { + return connection().send(CommandCode.Topic.GET_ALL.getValue(), payload).thenApply(response -> { try { List<Topic> topics = new ArrayList<>(); while (response.isReadable()) { @@ -106,7 +111,7 @@ public class TopicsTcpClient implements TopicsClient { payload.writeByte(replicationFactor.orElse((short) 0)); payload.writeBytes(BytesSerializer.toBytes(name)); - return connection.send(CommandCode.Topic.CREATE.getValue(), payload).thenApply(response -> { + return connection().send(CommandCode.Topic.CREATE.getValue(), payload).thenApply(response -> { try { return BytesDeserializer.readTopicDetails(response); } finally { @@ -134,7 +139,9 @@ public class TopicsTcpClient implements TopicsClient { payload.writeByte(replicationFactor.orElse((short) 0)); payload.writeBytes(BytesSerializer.toBytes(name)); - return connection.send(CommandCode.Topic.UPDATE.getValue(), payload).thenAccept(response -> response.release()); + return connection() + .send(CommandCode.Topic.UPDATE.getValue(), payload) + .thenAccept(response -> response.release()); } @Override @@ -143,6 +150,8 @@ public class TopicsTcpClient implements TopicsClient { payload.writeBytes(toBytes(streamId)); payload.writeBytes(toBytes(topicId)); - return connection.send(CommandCode.Topic.DELETE.getValue(), payload).thenAccept(response -> response.release()); + return connection() + .send(CommandCode.Topic.DELETE.getValue(), payload) + .thenAccept(response -> response.release()); } } diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/UsersTcpClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/UsersTcpClient.java index 632f45d6d..6ed7c3bef 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/UsersTcpClient.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/UsersTcpClient.java @@ -36,6 +36,7 @@ import org.slf4j.LoggerFactory; import java.util.List; import java.util.Optional; import java.util.concurrent.CompletableFuture; +import java.util.function.Supplier; import static org.apache.iggy.serde.BytesSerializer.toBytes; @@ -45,22 +46,32 @@ import static org.apache.iggy.serde.BytesSerializer.toBytes; public class UsersTcpClient implements UsersClient { private static final Logger log = LoggerFactory.getLogger(UsersTcpClient.class); - private final AsyncTcpConnection connection; + private final Supplier<AsyncTcpConnection> connectionSupplier; + private final LoginRedirectionHook redirectionHook; - public UsersTcpClient(AsyncTcpConnection connection) { - this.connection = connection; + public UsersTcpClient(Supplier<AsyncTcpConnection> connectionSupplier) { + this(connectionSupplier, LoginRedirectionHook.NONE); + } + + UsersTcpClient(Supplier<AsyncTcpConnection> connectionSupplier, LoginRedirectionHook redirectionHook) { + this.connectionSupplier = connectionSupplier; + this.redirectionHook = redirectionHook; + } + + private AsyncTcpConnection connection() { + return connectionSupplier.get(); } @Override public CompletableFuture<Optional<UserInfoDetails>> getUser(UserId userId) { var payload = toBytes(userId); - return connection.exchangeForOptional(CommandCode.User.GET, payload, BytesDeserializer::readUserInfoDetails); + return connection().exchangeForOptional(CommandCode.User.GET, payload, BytesDeserializer::readUserInfoDetails); } @Override public CompletableFuture<List<UserInfo>> getUsers() { var payload = Unpooled.EMPTY_BUFFER; - return connection.exchangeForList(CommandCode.User.GET_ALL, payload, BytesDeserializer::readUserInfo); + return connection().exchangeForList(CommandCode.User.GET_ALL, payload, BytesDeserializer::readUserInfo); } @Override @@ -79,13 +90,13 @@ public class UsersTcpClient implements UsersClient { }, () -> payload.writeByte(0)); - return connection.exchangeForEntity(CommandCode.User.CREATE, payload, BytesDeserializer::readUserInfoDetails); + return connection().exchangeForEntity(CommandCode.User.CREATE, payload, BytesDeserializer::readUserInfoDetails); } @Override public CompletableFuture<Void> deleteUser(UserId userId) { var payload = toBytes(userId); - return connection.sendAndRelease(CommandCode.User.DELETE, payload); + return connection().sendAndRelease(CommandCode.User.DELETE, payload); } @Override @@ -104,7 +115,7 @@ public class UsersTcpClient implements UsersClient { }, () -> payload.writeByte(0)); - return connection.sendAndRelease(CommandCode.User.UPDATE, payload); + return connection().sendAndRelease(CommandCode.User.UPDATE, payload); } @Override @@ -120,7 +131,7 @@ public class UsersTcpClient implements UsersClient { }, () -> payload.writeByte(0)); - return connection.sendAndRelease(CommandCode.User.UPDATE_PERMISSIONS, payload); + return connection().sendAndRelease(CommandCode.User.UPDATE_PERMISSIONS, payload); } @Override @@ -129,11 +140,17 @@ public class UsersTcpClient implements UsersClient { payload.writeBytes(toBytes(currentPassword)); payload.writeBytes(toBytes(newPassword)); - return connection.sendAndRelease(CommandCode.User.CHANGE_PASSWORD, payload); + return connection().sendAndRelease(CommandCode.User.CHANGE_PASSWORD, payload); } @Override public CompletableFuture<IdentityInfo> login(String username, String password) { + return loginWithoutRedirect(username, password).thenCompose(identity -> redirectionHook + .afterLogin(() -> loginWithoutRedirect(username, password)) + .thenApply(redirectedIdentity -> redirectedIdentity != null ? redirectedIdentity : identity)); + } + + private CompletableFuture<IdentityInfo> loginWithoutRedirect(String username, String password) { String version = IggyVersion.getInstance().getUserAgent(); String context = IggyVersion.getInstance().toString(); @@ -150,9 +167,8 @@ public class UsersTcpClient implements UsersClient { log.debug("Logging in user: {}", username); - return connection.send(CommandCode.User.LOGIN.getValue(), payload).thenApply(response -> { + return connection().send(CommandCode.User.LOGIN.getValue(), payload).thenApply(response -> { try { - // Read the user ID from response (4-byte unsigned int LE) var userId = response.readUnsignedIntLE(); return new IdentityInfo(userId, Optional.empty()); } finally { @@ -167,7 +183,7 @@ public class UsersTcpClient implements UsersClient { log.debug("Logging out"); - return connection.send(CommandCode.User.LOGOUT.getValue(), payload).thenAccept(response -> { + return connection().send(CommandCode.User.LOGOUT.getValue(), payload).thenAccept(response -> { response.release(); log.debug("Logged out successfully"); }); diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/SystemClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/SystemClient.java index ecf29e9aa..abb7b4260 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/SystemClient.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/SystemClient.java @@ -19,6 +19,7 @@ package org.apache.iggy.client.blocking; +import org.apache.iggy.cluster.ClusterMetadata; import org.apache.iggy.system.ClientInfo; import org.apache.iggy.system.ClientInfoDetails; import org.apache.iggy.system.Stats; @@ -29,6 +30,8 @@ public interface SystemClient { Stats getStats(); + ClusterMetadata getClusterMetadata(); + ClientInfoDetails getMe(); ClientInfoDetails getClient(Long clientId); diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/http/SystemHttpClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/http/SystemHttpClient.java index 3507919dc..d238c32f0 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/http/SystemHttpClient.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/http/SystemHttpClient.java @@ -20,6 +20,7 @@ package org.apache.iggy.client.blocking.http; import org.apache.iggy.client.blocking.SystemClient; +import org.apache.iggy.cluster.ClusterMetadata; import org.apache.iggy.exception.IggyOperationNotSupportedException; import org.apache.iggy.system.ClientInfo; import org.apache.iggy.system.ClientInfoDetails; @@ -31,6 +32,7 @@ import java.util.List; class SystemHttpClient implements SystemClient { private static final String STATS = "/stats"; + private static final String CLUSTER_METADATA = "/cluster/metadata"; private static final String CLIENTS = "/clients"; private static final String PING = "/ping"; private final InternalHttpClient httpClient; @@ -45,6 +47,12 @@ class SystemHttpClient implements SystemClient { return httpClient.execute(request, Stats.class); } + @Override + public ClusterMetadata getClusterMetadata() { + var request = httpClient.prepareGetRequest(CLUSTER_METADATA); + return httpClient.execute(request, ClusterMetadata.class); + } + @Override public ClientInfoDetails getMe() { throw new IggyOperationNotSupportedException("getMe", "HTTP"); diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/tcp/IggyTcpClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/tcp/IggyTcpClient.java index 2bf5eecde..8a60c27bc 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/tcp/IggyTcpClient.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/tcp/IggyTcpClient.java @@ -19,6 +19,7 @@ package org.apache.iggy.client.blocking.tcp; +import org.apache.iggy.client.ConnectionInfo; import org.apache.iggy.client.async.tcp.AsyncIggyTcpClient; import org.apache.iggy.client.blocking.ConsumerGroupsClient; import org.apache.iggy.client.blocking.ConsumerOffsetsClient; @@ -93,6 +94,17 @@ public class IggyTcpClient implements IggyBaseClient, Closeable { return FutureUtil.resolve(asyncClient.login()); } + /** + * Returns the server address this client currently targets. Leader + * redirection can change it after login, so it may differ from the + * address the client was built with. + * + * @return the current {@link ConnectionInfo} + */ + public ConnectionInfo getConnectionInfo() { + return asyncClient.getConnectionInfo(); + } + /** {@inheritDoc} */ @Override public byte[] sendBinaryRequest(int code, byte[] payload) { diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/tcp/SystemTcpClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/tcp/SystemTcpClient.java index 1d1df7fa4..4d00cfa14 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/tcp/SystemTcpClient.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/tcp/SystemTcpClient.java @@ -20,6 +20,7 @@ package org.apache.iggy.client.blocking.tcp; import org.apache.iggy.client.blocking.SystemClient; +import org.apache.iggy.cluster.ClusterMetadata; import org.apache.iggy.system.ClientInfo; import org.apache.iggy.system.ClientInfoDetails; import org.apache.iggy.system.Stats; @@ -39,6 +40,11 @@ final class SystemTcpClient implements SystemClient { return FutureUtil.resolve(delegate.getStats()); } + @Override + public ClusterMetadata getClusterMetadata() { + return FutureUtil.resolve(delegate.getClusterMetadata()); + } + @Override public ClientInfoDetails getMe() { return FutureUtil.resolve(delegate.getMe()); diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/SystemClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/cluster/ClusterMetadata.java similarity index 69% copy from foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/SystemClient.java copy to foreign/java/java-sdk/src/main/java/org/apache/iggy/cluster/ClusterMetadata.java index ecf29e9aa..757108c61 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/SystemClient.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/cluster/ClusterMetadata.java @@ -17,23 +17,17 @@ * under the License. */ -package org.apache.iggy.client.blocking; - -import org.apache.iggy.system.ClientInfo; -import org.apache.iggy.system.ClientInfoDetails; -import org.apache.iggy.system.Stats; +package org.apache.iggy.cluster; import java.util.List; -public interface SystemClient { - - Stats getStats(); - - ClientInfoDetails getMe(); - - ClientInfoDetails getClient(Long clientId); - - List<ClientInfo> getClients(); - - String ping(); -} +/** + * Cluster metadata reported by the server. + * + * <p>A server running without clustering reports the cluster name + * {@code "single-node"} with exactly one node. + * + * @param name the cluster name + * @param nodes the cluster nodes + */ +public record ClusterMetadata(String name, List<ClusterNode> nodes) {} diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/SystemClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/cluster/ClusterNode.java similarity index 65% copy from foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/SystemClient.java copy to foreign/java/java-sdk/src/main/java/org/apache/iggy/cluster/ClusterNode.java index ecf29e9aa..1114fca88 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/SystemClient.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/cluster/ClusterNode.java @@ -17,23 +17,16 @@ * under the License. */ -package org.apache.iggy.client.blocking; +package org.apache.iggy.cluster; -import org.apache.iggy.system.ClientInfo; -import org.apache.iggy.system.ClientInfoDetails; -import org.apache.iggy.system.Stats; - -import java.util.List; - -public interface SystemClient { - - Stats getStats(); - - ClientInfoDetails getMe(); - - ClientInfoDetails getClient(Long clientId); - - List<ClientInfo> getClients(); - - String ping(); -} +/** + * A single node in the cluster roster. + * + * @param name the node name + * @param ip the node IP address or hostname + * @param endpoints the per-transport ports of the node + * @param role the node role + * @param status the node status + */ +public record ClusterNode( + String name, String ip, TransportEndpoints endpoints, ClusterNodeRole role, ClusterNodeStatus status) {} diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/SystemClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/cluster/ClusterNodeRole.java similarity index 51% copy from foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/SystemClient.java copy to foreign/java/java-sdk/src/main/java/org/apache/iggy/cluster/ClusterNodeRole.java index ecf29e9aa..d95f7532a 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/SystemClient.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/cluster/ClusterNodeRole.java @@ -17,23 +17,36 @@ * under the License. */ -package org.apache.iggy.client.blocking; +package org.apache.iggy.cluster; -import org.apache.iggy.system.ClientInfo; -import org.apache.iggy.system.ClientInfoDetails; -import org.apache.iggy.system.Stats; +import com.fasterxml.jackson.annotation.JsonProperty; +import org.apache.iggy.exception.IggyMalformedResponseException; -import java.util.List; - -public interface SystemClient { - - Stats getStats(); - - ClientInfoDetails getMe(); - - ClientInfoDetails getClient(Long clientId); - - List<ClientInfo> getClients(); - - String ping(); +/** + * Role of a node in the cluster. + */ +public enum ClusterNodeRole { + @JsonProperty("leader") + Leader(0), + @JsonProperty("follower") + Follower(1); + + private final int code; + + ClusterNodeRole(int code) { + this.code = code; + } + + public static ClusterNodeRole fromCode(int code) { + for (ClusterNodeRole role : ClusterNodeRole.values()) { + if (role.code == code) { + return role; + } + } + throw new IggyMalformedResponseException("Invalid cluster node role: " + code); + } + + public int asCode() { + return code; + } } diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/cluster/ClusterNodeStatus.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/cluster/ClusterNodeStatus.java new file mode 100644 index 000000000..be115c5b6 --- /dev/null +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/cluster/ClusterNodeStatus.java @@ -0,0 +1,58 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iggy.cluster; + +import com.fasterxml.jackson.annotation.JsonProperty; +import org.apache.iggy.exception.IggyMalformedResponseException; + +/** + * Status of a node in the cluster. + */ +public enum ClusterNodeStatus { + @JsonProperty("healthy") + Healthy(0), + @JsonProperty("starting") + Starting(1), + @JsonProperty("stopping") + Stopping(2), + @JsonProperty("unreachable") + Unreachable(3), + @JsonProperty("maintenance") + Maintenance(4); + + private final int code; + + ClusterNodeStatus(int code) { + this.code = code; + } + + public static ClusterNodeStatus fromCode(int code) { + for (ClusterNodeStatus status : ClusterNodeStatus.values()) { + if (status.code == code) { + return status; + } + } + throw new IggyMalformedResponseException("Invalid cluster node status: " + code); + } + + public int asCode() { + return code; + } +} diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/SystemClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/cluster/TransportEndpoints.java similarity index 67% copy from foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/SystemClient.java copy to foreign/java/java-sdk/src/main/java/org/apache/iggy/cluster/TransportEndpoints.java index ecf29e9aa..4a4a68ec4 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/SystemClient.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/cluster/TransportEndpoints.java @@ -17,23 +17,15 @@ * under the License. */ -package org.apache.iggy.client.blocking; +package org.apache.iggy.cluster; -import org.apache.iggy.system.ClientInfo; -import org.apache.iggy.system.ClientInfoDetails; -import org.apache.iggy.system.Stats; - -import java.util.List; - -public interface SystemClient { - - Stats getStats(); - - ClientInfoDetails getMe(); - - ClientInfoDetails getClient(Long clientId); - - List<ClientInfo> getClients(); - - String ping(); -} +/** + * Per-transport ports of a cluster node. A port of {@code 0} means the + * transport is disabled on that node. + * + * @param tcp the TCP port + * @param quic the QUIC port + * @param http the HTTP port + * @param websocket the WebSocket port + */ +public record TransportEndpoints(int tcp, int quic, int http, int websocket) {} diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/SystemClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/cluster/package-info.java similarity index 67% copy from foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/SystemClient.java copy to foreign/java/java-sdk/src/main/java/org/apache/iggy/cluster/package-info.java index ecf29e9aa..c3385a3f6 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/SystemClient.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/cluster/package-info.java @@ -17,23 +17,7 @@ * under the License. */ -package org.apache.iggy.client.blocking; +@NonNullApi +package org.apache.iggy.cluster; -import org.apache.iggy.system.ClientInfo; -import org.apache.iggy.system.ClientInfoDetails; -import org.apache.iggy.system.Stats; - -import java.util.List; - -public interface SystemClient { - - Stats getStats(); - - ClientInfoDetails getMe(); - - ClientInfoDetails getClient(Long clientId); - - List<ClientInfo> getClients(); - - String ping(); -} +import org.apache.iggy.NonNullApi; diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/serde/BytesDeserializer.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/serde/BytesDeserializer.java index 7b4328b0c..67afa0852 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/serde/BytesDeserializer.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/serde/BytesDeserializer.java @@ -21,10 +21,16 @@ package org.apache.iggy.serde; import io.netty.buffer.ByteBuf; import org.apache.commons.lang3.ArrayUtils; +import org.apache.iggy.cluster.ClusterMetadata; +import org.apache.iggy.cluster.ClusterNode; +import org.apache.iggy.cluster.ClusterNodeRole; +import org.apache.iggy.cluster.ClusterNodeStatus; +import org.apache.iggy.cluster.TransportEndpoints; import org.apache.iggy.consumergroup.ConsumerGroup; import org.apache.iggy.consumergroup.ConsumerGroupDetails; import org.apache.iggy.consumergroup.ConsumerGroupMember; import org.apache.iggy.consumeroffset.ConsumerOffsetInfo; +import org.apache.iggy.exception.IggyMalformedResponseException; import org.apache.iggy.message.BytesMessageId; import org.apache.iggy.message.HeaderKey; import org.apache.iggy.message.HeaderKind; @@ -68,6 +74,8 @@ import java.util.Optional; */ public final class BytesDeserializer { + private static final int MIN_CLUSTER_NODE_BYTES = 18; + private BytesDeserializer() {} public static StreamBase readStreamBase(ByteBuf response) { @@ -349,6 +357,33 @@ public final class BytesDeserializer { return new ConsumerGroupInfo(streamId, topicId, groupId); } + public static ClusterMetadata readClusterMetadata(ByteBuf response) { + var name = readU32PrefixedString(response, "cluster name"); + var nodesCount = response.readUnsignedIntLE(); + if (nodesCount > response.readableBytes() / MIN_CLUSTER_NODE_BYTES) { + throw new IggyMalformedResponseException("Cluster nodes count " + nodesCount + + " exceeds remaining payload of " + response.readableBytes() + " bytes"); + } + List<ClusterNode> nodes = new ArrayList<>(toInt(nodesCount)); + for (int i = 0; i < nodesCount; i++) { + nodes.add(readClusterNode(response)); + } + return new ClusterMetadata(name, nodes); + } + + public static ClusterNode readClusterNode(ByteBuf response) { + var name = readU32PrefixedString(response, "cluster node name"); + var ip = readU32PrefixedString(response, "cluster node ip"); + var tcpPort = response.readUnsignedShortLE(); + var quicPort = response.readUnsignedShortLE(); + var httpPort = response.readUnsignedShortLE(); + var websocketPort = response.readUnsignedShortLE(); + var role = ClusterNodeRole.fromCode(response.readUnsignedByte()); + var status = ClusterNodeStatus.fromCode(response.readUnsignedByte()); + return new ClusterNode( + name, ip, new TransportEndpoints(tcpPort, quicPort, httpPort, websocketPort), role, status); + } + public static UserInfoDetails readUserInfoDetails(ByteBuf response) { var userInfo = readUserInfo(response); @@ -448,6 +483,18 @@ public final class BytesDeserializer { return new PersonalAccessTokenInfo(name, expiryOptional); } + private static String readU32PrefixedString(ByteBuf buffer, String field) { + if (buffer.readableBytes() < Integer.BYTES) { + throw new IggyMalformedResponseException("Missing length prefix for " + field); + } + var length = buffer.readUnsignedIntLE(); + if (length > buffer.readableBytes()) { + throw new IggyMalformedResponseException("Length " + length + " for " + field + + " exceeds remaining payload of " + buffer.readableBytes() + " bytes"); + } + return buffer.readCharSequence(toInt(length), StandardCharsets.UTF_8).toString(); + } + static BigInteger readU64AsBigInteger(ByteBuf buffer) { var bytesArray = new byte[8]; buffer.readBytes(bytesArray, 0, 8); diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/serde/CommandCode.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/serde/CommandCode.java index 236394707..337494d18 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/serde/CommandCode.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/serde/CommandCode.java @@ -30,6 +30,7 @@ public interface CommandCode { enum System implements CommandCode { PING(1), GET_STATS(10), + GET_CLUSTER_METADATA(12), GET_ME(20), GET_CLIENT(21), GET_ALL_CLIENTS(22); diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/IggyResponseHandlerTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/IggyResponseHandlerTest.java new file mode 100644 index 000000000..fe542542f --- /dev/null +++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/IggyResponseHandlerTest.java @@ -0,0 +1,60 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iggy.client.async.tcp; + +import io.netty.buffer.ByteBuf; +import io.netty.channel.embedded.EmbeddedChannel; +import org.apache.iggy.exception.IggyConnectionException; +import org.junit.jupiter.api.Test; + +import java.util.concurrent.CompletableFuture; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class IggyResponseHandlerTest { + + @Test + void shouldFailPendingRequestsWhenChannelBecomesInactive() { + var handler = new AsyncTcpConnection.IggyResponseHandler(); + var channel = new EmbeddedChannel(handler); + CompletableFuture<ByteBuf> pending = new CompletableFuture<>(); + handler.enqueueRequest(pending); + + channel.close(); + + assertThat(pending).isCompletedExceptionally(); + assertThatThrownBy(pending::join).hasCauseInstanceOf(IggyConnectionException.class); + } + + @Test + void shouldFailPendingRequestsOnPipelineException() { + var handler = new AsyncTcpConnection.IggyResponseHandler(); + var channel = new EmbeddedChannel(handler); + CompletableFuture<ByteBuf> pending = new CompletableFuture<>(); + handler.enqueueRequest(pending); + + var failure = new IllegalStateException("broken pipe"); + channel.pipeline().fireExceptionCaught(failure); + + assertThat(pending).isCompletedExceptionally(); + assertThatThrownBy(pending::join).hasCause(failure); + } +} diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/LeaderAwarenessTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/LeaderAwarenessTest.java new file mode 100644 index 000000000..00540cd23 --- /dev/null +++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/LeaderAwarenessTest.java @@ -0,0 +1,328 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iggy.client.async.tcp; + +import org.apache.iggy.client.ConnectionInfo; +import org.apache.iggy.client.async.tcp.LeaderAwareness.LeaderCheck; +import org.apache.iggy.cluster.ClusterMetadata; +import org.apache.iggy.cluster.ClusterNode; +import org.apache.iggy.cluster.ClusterNodeRole; +import org.apache.iggy.cluster.ClusterNodeStatus; +import org.apache.iggy.cluster.TransportEndpoints; +import org.apache.iggy.exception.IggyConnectionException; +import org.junit.jupiter.api.Nested; +import org.junit.jupiter.api.Test; + +import java.time.Duration; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Supplier; + +import static org.assertj.core.api.Assertions.assertThat; + +class LeaderAwarenessTest { + + private static ClusterNode node( + String name, String ip, int tcpPort, ClusterNodeRole role, ClusterNodeStatus status) { + return new ClusterNode(name, ip, new TransportEndpoints(tcpPort, 8080, 3000, 8070), role, status); + } + + private static ClusterMetadata cluster(ClusterNode... nodes) { + return new ClusterMetadata("test-cluster", List.of(nodes)); + } + + private static ClusterMetadata leaderlessCluster() { + return cluster( + node("node-0", "iggy-0", 8091, ClusterNodeRole.Follower, ClusterNodeStatus.Healthy), + node("node-1", "iggy-1", 8092, ClusterNodeRole.Follower, ClusterNodeStatus.Healthy)); + } + + private static ClusterMetadata clusterWithLeaderElsewhere() { + return cluster( + node("leader-node", "iggy-leader", 8091, ClusterNodeRole.Leader, ClusterNodeStatus.Healthy), + node("follower-node", "iggy-follower", 8092, ClusterNodeRole.Follower, ClusterNodeStatus.Healthy)); + } + + @Nested + class CheckLeader { + + @Test + void shouldStayForSingleNodeCluster() { + var metadata = + cluster(node("iggy-node", "other-host", 8090, ClusterNodeRole.Leader, ClusterNodeStatus.Healthy)); + + var check = LeaderAwareness.checkLeader(metadata, new ConnectionInfo("localhost", 8090)); + + assertThat(check).isInstanceOf(LeaderCheck.Stay.class); + } + + @Test + void shouldRedirectToHealthyLeaderOnAnotherNode() { + var check = LeaderAwareness.checkLeader( + clusterWithLeaderElsewhere(), new ConnectionInfo("iggy-follower", 8092)); + + assertThat(check).isEqualTo(new LeaderCheck.Redirect(new ConnectionInfo("iggy-leader", 8091))); + } + + @Test + void shouldStayWhenAlreadyOnLeader() { + var check = + LeaderAwareness.checkLeader(clusterWithLeaderElsewhere(), new ConnectionInfo("iggy-leader", 8091)); + + assertThat(check).isInstanceOf(LeaderCheck.Stay.class); + } + + @Test + void shouldReportLeaderlessClusterAsNoLeader() { + var check = LeaderAwareness.checkLeader(leaderlessCluster(), new ConnectionInfo("iggy-1", 8092)); + + assertThat(check).isInstanceOf(LeaderCheck.NoLeader.class); + } + + @Test + void shouldReportUnhealthyLeaderAsNoLeader() { + var metadata = cluster( + node("leader-node", "iggy-leader", 8091, ClusterNodeRole.Leader, ClusterNodeStatus.Unreachable), + node("follower-node", "iggy-follower", 8092, ClusterNodeRole.Follower, ClusterNodeStatus.Healthy)); + + var check = LeaderAwareness.checkLeader(metadata, new ConnectionInfo("iggy-follower", 8092)); + + assertThat(check).isInstanceOf(LeaderCheck.NoLeader.class); + } + + @Test + void shouldStayWhenLeaderTcpTransportIsDisabled() { + var metadata = cluster( + node("leader-node", "iggy-leader", 0, ClusterNodeRole.Leader, ClusterNodeStatus.Healthy), + node("follower-node", "iggy-follower", 8092, ClusterNodeRole.Follower, ClusterNodeStatus.Healthy)); + + var check = LeaderAwareness.checkLeader(metadata, new ConnectionInfo("iggy-follower", 8092)); + + assertThat(check).isInstanceOf(LeaderCheck.Stay.class); + } + + @Test + void shouldTreatLocalhostAndLoopbackAsSameAddress() { + var metadata = cluster( + node("leader-node", "127.0.0.1", 8091, ClusterNodeRole.Leader, ClusterNodeStatus.Healthy), + node("follower-node", "127.0.0.1", 8092, ClusterNodeRole.Follower, ClusterNodeStatus.Healthy)); + + var check = LeaderAwareness.checkLeader(metadata, new ConnectionInfo("localhost", 8091)); + + assertThat(check).isInstanceOf(LeaderCheck.Stay.class); + } + + @Test + void shouldReportEmptyRosterAsNoLeader() { + var check = LeaderAwareness.checkLeader(cluster(), new ConnectionInfo("localhost", 8090)); + + assertThat(check).isInstanceOf(LeaderCheck.NoLeader.class); + } + } + + @Nested + class AddressComparison { + + private boolean isSameAddress(String host1, int port1, String host2, int port2) { + return LeaderAwareness.isSameAddress(new ConnectionInfo(host1, port1), new ConnectionInfo(host2, port2)); + } + + @Test + void shouldMatchIdenticalAddresses() { + assertThat(isSameAddress("127.0.0.1", 8090, "127.0.0.1", 8090)).isTrue(); + } + + @Test + void shouldMatchLocalhostAgainstLoopback() { + assertThat(isSameAddress("localhost", 8090, "127.0.0.1", 8090)).isTrue(); + } + + @Test + void shouldMatchIpv4LoopbackAgainstIpv6Loopback() { + assertThat(isSameAddress("127.0.0.1", 8090, "::1", 8090)).isTrue(); + } + + @Test + void shouldMatchCaseInsensitively() { + assertThat(isSameAddress("LOCALHOST", 8090, "127.0.0.1", 8090)).isTrue(); + assertThat(isSameAddress("Iggy-Leader", 8091, "iggy-leader", 8091)).isTrue(); + } + + @Test + void shouldMatchUnspecifiedIpv6AgainstLoopback() { + assertThat(isSameAddress("[::]", 8090, "[::1]", 8090)).isTrue(); + } + + @Test + void shouldMatchBracketedIpv6AgainstBareForm() { + assertThat(isSameAddress("[::1]", 8090, "::1", 8090)).isTrue(); + } + + @Test + void shouldRejectDifferentPorts() { + assertThat(isSameAddress("127.0.0.1", 8090, "127.0.0.1", 8091)).isFalse(); + } + + @Test + void shouldRejectDifferentHosts() { + assertThat(isSameAddress("192.168.1.1", 8090, "127.0.0.1", 8090)).isFalse(); + } + + @Test + void shouldNotRewriteHostnamesEmbeddingLocalhost() { + // .invalid never resolves (RFC 2606), so only mangling the names + // into each other could make these compare equal + assertThat(isSameAddress("mylocalhost.invalid", 8090, "my127.0.0.1.invalid", 8090)) + .isFalse(); + } + } + + @Nested + class FindLeaderElsewhere { + + private static final Duration BUDGET = Duration.ofSeconds(2); + private static final Duration INTERVAL = Duration.ofMillis(10); + + private final ConnectionInfo currentTarget = new ConnectionInfo("iggy-follower", 8092); + + private Optional<ConnectionInfo> findLeader(Supplier<CompletableFuture<ClusterMetadata>> fetch) { + return LeaderAwareness.findLeaderElsewhere(fetch, currentTarget, BUDGET, INTERVAL) + .orTimeout(30, TimeUnit.SECONDS) + .join(); + } + + @Test + void shouldRedirectWithoutPollingWhenLeaderIsKnown() { + var fetchCount = new AtomicInteger(); + + var leader = findLeader(() -> { + fetchCount.incrementAndGet(); + return CompletableFuture.completedFuture(clusterWithLeaderElsewhere()); + }); + + assertThat(leader).contains(new ConnectionInfo("iggy-leader", 8091)); + assertThat(fetchCount).hasValue(1); + } + + @Test + void shouldPollThroughLeaderlessElectionUntilLeaderAppears() { + var fetchCount = new AtomicInteger(); + + var leader = findLeader(() -> CompletableFuture.completedFuture( + fetchCount.incrementAndGet() < 3 ? leaderlessCluster() : clusterWithLeaderElsewhere())); + + assertThat(leader).contains(new ConnectionInfo("iggy-leader", 8091)); + assertThat(fetchCount).hasValue(3); + } + + @Test + void shouldGiveUpOnLeaderlessClusterAfterBudget() { + var fetchCount = new AtomicInteger(); + + var leader = LeaderAwareness.findLeaderElsewhere( + () -> { + fetchCount.incrementAndGet(); + return CompletableFuture.completedFuture(leaderlessCluster()); + }, + currentTarget, + Duration.ofMillis(100), + INTERVAL) + .orTimeout(30, TimeUnit.SECONDS) + .join(); + + assertThat(leader).isEmpty(); + assertThat(fetchCount.get()).isGreaterThan(1); + } + + @Test + void shouldGiveUpImmediatelyWhenMetadataFetchFails() { + var fetchCount = new AtomicInteger(); + + var leader = findLeader(() -> { + fetchCount.incrementAndGet(); + return CompletableFuture.failedFuture(new IggyConnectionException("connection lost")); + }); + + assertThat(leader).isEmpty(); + assertThat(fetchCount).hasValue(1); + } + + @Test + void shouldGiveUpWhenMetadataFetchThrowsSynchronously() { + var leader = findLeader(() -> { + throw new IllegalStateException("no connection"); + }); + + assertThat(leader).isEmpty(); + } + + @Test + void shouldStayWithoutPollingWhenAlreadyOnLeader() { + var fetchCount = new AtomicInteger(); + + var leader = LeaderAwareness.findLeaderElsewhere( + () -> { + fetchCount.incrementAndGet(); + return CompletableFuture.completedFuture(clusterWithLeaderElsewhere()); + }, + new ConnectionInfo("iggy-leader", 8091), + BUDGET, + INTERVAL) + .orTimeout(30, TimeUnit.SECONDS) + .join(); + + assertThat(leader).isEmpty(); + assertThat(fetchCount).hasValue(1); + } + } + + @Nested + class RedirectionState { + + @Test + void shouldCapConsecutiveRedirectsAtThree() { + var state = new LeaderAwareness.LeaderRedirectionState(); + assertThat(state.canRedirect()).isTrue(); + + state.recordRedirect(); + state.recordRedirect(); + assertThat(state.canRedirect()).isTrue(); + + state.recordRedirect(); + assertThat(state.canRedirect()).isFalse(); + } + + @Test + void shouldGrantFreshBudgetToEachCheck() { + var exhausted = new LeaderAwareness.LeaderRedirectionState(); + exhausted.recordRedirect(); + exhausted.recordRedirect(); + exhausted.recordRedirect(); + assertThat(exhausted.canRedirect()).isFalse(); + + var nextCheck = new LeaderAwareness.LeaderRedirectionState(); + + assertThat(nextCheck.canRedirect()).isTrue(); + } + } +} diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/SystemClientBaseTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/SystemClientBaseTest.java index 01d55315b..550b69b40 100644 --- a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/SystemClientBaseTest.java +++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/SystemClientBaseTest.java @@ -19,6 +19,8 @@ package org.apache.iggy.client.blocking; +import org.apache.iggy.cluster.ClusterNodeRole; +import org.apache.iggy.cluster.ClusterNodeStatus; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -49,6 +51,22 @@ public abstract class SystemClientBaseTest extends IntegrationTest { assertThat(stats.totalDiskSpace()).isNotNull(); } + @Test + void shouldGetClusterMetadataForSingleNode() { + // when + var metadata = systemClient.getClusterMetadata(); + + // then + assertThat(metadata).isNotNull(); + assertThat(metadata.name()).isEqualTo("single-node"); + assertThat(metadata.nodes()).hasSize(1); + var node = metadata.nodes().get(0); + assertThat(node.name()).isEqualTo("iggy-node"); + assertThat(node.role()).isEqualTo(ClusterNodeRole.Leader); + assertThat(node.status()).isEqualTo(ClusterNodeStatus.Healthy); + assertThat(node.endpoints().tcp()).isPositive(); + } + @Test void shouldPing() { // when diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/tcp/SystemTcpClientTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/tcp/SystemTcpClientTest.java index db00a876f..6337a502b 100644 --- a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/tcp/SystemTcpClientTest.java +++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/tcp/SystemTcpClientTest.java @@ -60,4 +60,14 @@ class SystemTcpClientTest extends SystemClientBaseTest { assertThat(clients).isNotNull(); assertThat(clients.size()).isGreaterThanOrEqualTo(1); // At least our connection } + + @Test + void shouldStayOnSingleNodeAfterLogin() { + // when + var connectionInfo = ((IggyTcpClient) client).getConnectionInfo(); + + // then + assertThat(connectionInfo.host()).isEqualTo(serverHost()); + assertThat(connectionInfo.port()).isEqualTo(serverTcpPort()); + } } diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/serde/BytesDeserializerTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/serde/BytesDeserializerTest.java index f592ce5a5..637456ab1 100644 --- a/foreign/java/java-sdk/src/test/java/org/apache/iggy/serde/BytesDeserializerTest.java +++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/serde/BytesDeserializerTest.java @@ -22,6 +22,10 @@ package org.apache.iggy.serde; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import org.apache.commons.lang3.ArrayUtils; +import org.apache.iggy.cluster.ClusterNodeRole; +import org.apache.iggy.cluster.ClusterNodeStatus; +import org.apache.iggy.cluster.TransportEndpoints; +import org.apache.iggy.exception.IggyMalformedResponseException; import org.apache.iggy.message.HeaderKey; import org.apache.iggy.message.HeaderKind; import org.apache.iggy.system.CacheMetricsKey; @@ -36,6 +40,7 @@ import java.util.HexFormat; import static org.apache.iggy.serde.BytesDeserializer.readClientInfo; import static org.apache.iggy.serde.BytesDeserializer.readClientInfoDetails; +import static org.apache.iggy.serde.BytesDeserializer.readClusterMetadata; import static org.apache.iggy.serde.BytesDeserializer.readConsumerGroup; import static org.apache.iggy.serde.BytesDeserializer.readConsumerGroupDetails; import static org.apache.iggy.serde.BytesDeserializer.readConsumerGroupInfo; @@ -59,6 +64,7 @@ import static org.apache.iggy.serde.BytesDeserializer.readU64AsBigInteger; import static org.apache.iggy.serde.BytesDeserializer.readUserInfo; import static org.apache.iggy.serde.BytesDeserializer.readUserInfoDetails; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; class BytesDeserializerTest { @@ -889,4 +895,175 @@ class BytesDeserializerTest { assertThat(tokenInfo.expiryAt()).isEmpty(); } } + + @Nested + class ClusterMetadataDeserialization { + + private void writeU32PrefixedString(ByteBuf buffer, String value) { + byte[] bytes = value.getBytes(StandardCharsets.UTF_8); + buffer.writeIntLE(bytes.length); + buffer.writeBytes(bytes); + } + + private void writeClusterNode(ByteBuf buffer, String name, String ip, int tcpPort, int role, int status) { + writeU32PrefixedString(buffer, name); + writeU32PrefixedString(buffer, ip); + buffer.writeShortLE(tcpPort); + buffer.writeShortLE(8080); // quic + buffer.writeShortLE(3000); // http + buffer.writeShortLE(8070); // websocket + buffer.writeByte(role); + buffer.writeByte(status); + } + + private ByteBuf twoNodeCluster() { + ByteBuf buffer = Unpooled.buffer(); + writeU32PrefixedString(buffer, "test-cluster"); + buffer.writeIntLE(2); + writeClusterNode(buffer, "leader-node", "iggy-leader", 8091, 0, 0); + writeClusterNode(buffer, "follower-node", "iggy-follower", 8092, 1, 3); + return buffer; + } + + @Test + void shouldDeserializeMultiNodeCluster() { + // given + ByteBuf buffer = twoNodeCluster(); + + // when + var metadata = readClusterMetadata(buffer); + + // then + assertThat(metadata.name()).isEqualTo("test-cluster"); + assertThat(metadata.nodes()).hasSize(2); + var leader = metadata.nodes().get(0); + assertThat(leader.name()).isEqualTo("leader-node"); + assertThat(leader.ip()).isEqualTo("iggy-leader"); + assertThat(leader.endpoints()).isEqualTo(new TransportEndpoints(8091, 8080, 3000, 8070)); + assertThat(leader.role()).isEqualTo(ClusterNodeRole.Leader); + assertThat(leader.status()).isEqualTo(ClusterNodeStatus.Healthy); + var follower = metadata.nodes().get(1); + assertThat(follower.name()).isEqualTo("follower-node"); + assertThat(follower.role()).isEqualTo(ClusterNodeRole.Follower); + assertThat(follower.status()).isEqualTo(ClusterNodeStatus.Unreachable); + assertThat(buffer.isReadable()).isFalse(); + } + + @Test + void shouldDeserializeClusterWithoutNodes() { + // given + ByteBuf buffer = Unpooled.buffer(); + writeU32PrefixedString(buffer, "empty-cluster"); + buffer.writeIntLE(0); + + // when + var metadata = readClusterMetadata(buffer); + + // then + assertThat(metadata.name()).isEqualTo("empty-cluster"); + assertThat(metadata.nodes()).isEmpty(); + } + + @Test + void shouldDeserializeClusterWithEmptyName() { + // given + ByteBuf buffer = Unpooled.buffer(); + writeU32PrefixedString(buffer, ""); + buffer.writeIntLE(1); + writeClusterNode(buffer, "iggy-node", "localhost", 8090, 0, 0); + + // when + var metadata = readClusterMetadata(buffer); + + // then + assertThat(metadata.name()).isEmpty(); + assertThat(metadata.nodes()).hasSize(1); + } + + @Test + void shouldDeserializePortZeroAsDisabledTransport() { + // given + ByteBuf buffer = Unpooled.buffer(); + writeU32PrefixedString(buffer, "test-cluster"); + buffer.writeIntLE(2); + writeClusterNode(buffer, "leader-node", "iggy-leader", 0, 0, 0); + writeClusterNode(buffer, "follower-node", "iggy-follower", 8092, 1, 0); + + // when + var metadata = readClusterMetadata(buffer); + + // then + assertThat(metadata.nodes().get(0).endpoints().tcp()).isZero(); + } + + @Test + void shouldFailOnTruncationAtEveryByte() { + // given + ByteBuf complete = twoNodeCluster(); + byte[] bytes = new byte[complete.readableBytes()]; + complete.getBytes(0, bytes); + + for (int length = 0; length < bytes.length; length++) { + ByteBuf truncated = Unpooled.wrappedBuffer(bytes, 0, length); + + // when / then + assertThatThrownBy(() -> readClusterMetadata(truncated)) + .as("truncated at byte %d", length) + .isInstanceOf(RuntimeException.class); + } + } + + @Test + void shouldFailOnBogusNodesCountWithoutAllocating() { + // given + ByteBuf buffer = Unpooled.buffer(); + writeU32PrefixedString(buffer, "test-cluster"); + buffer.writeIntLE((int) 0xFFFFFFFFL); // u32::MAX nodes + + // when / then + assertThatThrownBy(() -> readClusterMetadata(buffer)) + .isInstanceOf(IggyMalformedResponseException.class) + .hasMessageContaining("nodes count"); + } + + @Test + void shouldFailOnBogusStringLength() { + // given + ByteBuf buffer = Unpooled.buffer(); + buffer.writeIntLE((int) 0xFFFFFFFFL); // u32::MAX name length + + // when / then + assertThatThrownBy(() -> readClusterMetadata(buffer)).isInstanceOf(IggyMalformedResponseException.class); + } + + @Test + void shouldFailOnUnknownRole() { + // given + ByteBuf buffer = Unpooled.buffer(); + writeU32PrefixedString(buffer, "test-cluster"); + buffer.writeIntLE(2); + writeClusterNode(buffer, "leader-node", "iggy-leader", 8091, 2, 0); + writeClusterNode(buffer, "follower-node", "iggy-follower", 8092, 1, 0); + + // when / then + assertThatThrownBy(() -> readClusterMetadata(buffer)) + .isInstanceOf(IggyMalformedResponseException.class) + .hasMessageContaining("role"); + } + + @Test + void shouldFailOnUnknownStatus() { + // given — 5 maps to the Rust Unknown variant, which TryFrom rejects + ByteBuf buffer = Unpooled.buffer(); + writeU32PrefixedString(buffer, "test-cluster"); + buffer.writeIntLE(2); + writeClusterNode(buffer, "leader-node", "iggy-leader", 8091, 0, 5); + writeClusterNode(buffer, "follower-node", "iggy-follower", 8092, 1, 0); + + // when / then + assertThatThrownBy(() -> readClusterMetadata(buffer)) + .isInstanceOf(IggyMalformedResponseException.class) + .hasMessageContaining("status"); + } + } } diff --git a/scripts/run-bdd-tests.sh b/scripts/run-bdd-tests.sh index cb86d7b8b..cfc000796 100755 --- a/scripts/run-bdd-tests.sh +++ b/scripts/run-bdd-tests.sh @@ -119,7 +119,7 @@ run_suite(){ local svc="$1" emoji="$2" label="$3" if [ "$FEATURE" = "leader_redirection" ]; then case "$svc" in - rust-bdd|go-bdd|csharp-bdd) ;; + rust-bdd|go-bdd|csharp-bdd|java-bdd) ;; *) if [ "$SDK" = "all" ]; then log "⚠️ skipping ${svc%-bdd} (does not support ${FEATURE})"
