diff --git a/bdd/docker-compose.cluster.yml b/bdd/docker-compose.cluster.yml index 7f62796211..063daf0db4 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 8a103db894..3069546d52 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 0000000000..5dc9778b7d --- /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 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 leaderA = leaderFromMetadata(clientA); + Optional 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 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 5339adf068..f9eb048aea 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/ConnectionInfo.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/ConnectionInfo.java new file mode 100644 index 0000000000..7042c5fce1 --- /dev/null +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/ConnectionInfo.java @@ -0,0 +1,40 @@ +/* + * 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; + +/** + * 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 f5587142d7..2ac3f0b3a9 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 getStats(); + /** + * Gets cluster metadata asynchronously. + * + * @return A CompletableFuture containing the cluster roster + */ + CompletableFuture 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 0af9a619c2..5a78acdbaf 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 @@ 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.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.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 username; private final Optional password; private final Optional connectionTimeout; @@ -112,7 +115,12 @@ public class AsyncIggyTcpClient { private final Optional retryPolicy; private final boolean enableTls; private final Optional tlsCertificate; - private AsyncTcpConnection connection; + private final TcpConnectionPoolConfig poolConfig; + private final AtomicReference connection = new AtomicReference<>(); + private final AtomicReference> 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 AsyncIggyTcpClient(String host, int port) { RetryPolicy retryPolicy, boolean enableTls, Optional 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 AsyncIggyTcpClient(String host, int port) { 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 static AsyncIggyTcpClientBuilder builder() { * @return a {@link CompletableFuture} that completes when the connection is established */ public CompletableFuture 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 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 CompletableFuture sendBinaryRequest(int code, byte[] payload) { 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 ConsumerOffsetsClient consumerOffsets() { * @return a {@link CompletableFuture} that completes when all resources are released */ public CompletableFuture 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 checkLeaderAndRedirect(Supplier> reLogin) { + CompletableFuture gate = new CompletableFuture<>(); + CompletableFuture 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 redirectToLeader( + Supplier> 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> findLeaderElsewhere(ConnectionInfo currentTarget) { + SystemClient currentSystemClient = systemClient; + if (currentSystemClient == null) { + return CompletableFuture.completedFuture(Optional.empty()); + } + return LeaderAwareness.findLeaderElsewhere(currentSystemClient::getClusterMetadata, currentTarget); + } + + private CompletableFuture 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 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 c74c615d46..38eb30f05e 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 AsyncTcpConnection( int port, boolean enableTls, Optional tlsCertificate, - TCPConnectionPoolConfig poolConfig, + TcpConnectionPoolConfig poolConfig, Optional connectionTimeout) { - this.eventLoopGroup = new MultiThreadIoEventLoopGroup(NioIoHandler.newFactory()); - SslContext sslContext = null; if (enableTls) { try { @@ -99,6 +97,8 @@ public AsyncTcpConnection( } } + this.eventLoopGroup = new MultiThreadIoEventLoopGroup(NioIoHandler.newFactory()); + var bootstrap = new Bootstrap() .group(eventLoopGroup) .channel(NioSocketChannel.class) @@ -195,11 +195,12 @@ public CompletableFuture send(CommandCode commandCode, ByteBuf payload) public CompletableFuture send(int commandCode, ByteBuf payload) { captureLoginPayloadIfNeeded(commandCode, payload); CompletableFuture responseFuture = new CompletableFuture<>(); + CompletableFuture callerFuture = new CompletableFuture<>(); channelPool.acquire().addListener((FutureListener) f -> { if (!f.isSuccess()) { payload.release(); - responseFuture.completeExceptionally(mapAcquireException(f.cause())); + callerFuture.completeExceptionally(mapAcquireException(f.cause())); return; } @@ -210,9 +211,17 @@ public CompletableFuture send(int commandCode, ByteBuf payload) { && 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 authStep; @@ -242,7 +251,7 @@ public CompletableFuture send(int commandCode, ByteBuf payload) { }); }); - return responseFuture; + return callerFuture; } private static Throwable mapAcquireException(Throwable cause) { @@ -396,34 +405,48 @@ protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) { } } + @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 f; - while ((f = responseQueue.poll()) != null) { - f.completeExceptionally(cause); - } + failPendingRequests(cause); ctx.close(); } + + private void failPendingRequests(Throwable cause) { + CompletableFuture 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 long getAcquireTimeoutMillis() { 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 static final class Builder { 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 Builder setMaxConnections(int maxConnections) { 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 Builder setMaxPendingAcquires(int maxPendingAcquires) { 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 Builder setAcquireTimeoutMillis(long acquireTimeoutMillis) { return this; } - public TCPConnectionPoolConfig build() { + public TcpConnectionPoolConfig build() { if (this.maxConnections == 0) { this.maxConnections = DEFAULT_MAX_CONNECTION; } @@ -481,7 +504,7 @@ public TCPConnectionPoolConfig build() { 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 71133f1552..9b4561fc64 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.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 @@ public class ConsumerGroupsTcpClient implements ConsumerGroupsClient { private static final Logger log = LoggerFactory.getLogger(ConsumerGroupsTcpClient.class); - private final AsyncTcpConnection connection; + private final Supplier connectionSupplier; - public ConsumerGroupsTcpClient(AsyncTcpConnection connection) { - this.connection = connection; + public ConsumerGroupsTcpClient(Supplier connectionSupplier) { + this.connectionSupplier = connectionSupplier; + } + + private AsyncTcpConnection connection() { + return connectionSupplier.get(); } @Override @@ -58,7 +63,7 @@ public CompletableFuture> getConsumerGroup( 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 CompletableFuture> getConsumerGroups(StreamId streamI 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 CompletableFuture createConsumerGroup( 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 CompletableFuture deleteConsumerGroup(StreamId streamId, TopicId to 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 CompletableFuture joinConsumerGroup(StreamId streamId, TopicId topi 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 CompletableFuture leaveConsumerGroup(StreamId streamId, TopicId top 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 aa6c6e41c1..cb53a72e25 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 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 @@ public class ConsumerOffsetsTcpClient implements ConsumerOffsetsClient { private static final Logger log = LoggerFactory.getLogger(ConsumerOffsetsTcpClient.class); - private final AsyncTcpConnection connection; + private final Supplier connectionSupplier; - public ConsumerOffsetsTcpClient(AsyncTcpConnection connection) { - this.connection = connection; + public ConsumerOffsetsTcpClient(Supplier connectionSupplier) { + this.connectionSupplier = connectionSupplier; + } + + private AsyncTcpConnection connection() { + return connectionSupplier.get(); } @Override @@ -63,7 +68,7 @@ public CompletableFuture storeConsumerOffset( consumer, offset); - return connection + return connection() .send(CommandCode.ConsumerOffset.STORE.getValue(), payload) .thenAccept(response -> { response.release(); @@ -85,7 +90,7 @@ public CompletableFuture> getConsumerOffset( 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 0000000000..f541e9de7c --- /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> findLeaderElsewhere( + Supplier> fetchMetadata, ConnectionInfo currentTarget) { + return findLeaderElsewhere(fetchMetadata, currentTarget, LEADERLESS_WAIT_BUDGET, LEADERLESS_POLL_INTERVAL); + } + + static CompletableFuture> findLeaderElsewhere( + Supplier> fetchMetadata, + ConnectionInfo currentTarget, + Duration leaderlessWaitBudget, + Duration leaderlessPollInterval) { + long electionDeadlineNanos = System.nanoTime() + leaderlessWaitBudget.toNanos(); + return pollForLeader( + fetchMetadata, currentTarget, leaderlessWaitBudget, leaderlessPollInterval, electionDeadlineNanos); + } + + private static CompletableFuture> pollForLeader( + Supplier> fetchMetadata, + ConnectionInfo currentTarget, + Duration leaderlessWaitBudget, + Duration leaderlessPollInterval, + long electionDeadlineNanos) { + CompletableFuture fetched; + try { + fetched = fetchMetadata.get(); + } catch (RuntimeException fetchError) { + fetched = CompletableFuture.failedFuture(fetchError); + } + return fetched.>>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 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 0000000000..d1c269f9ef --- /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 afterLogin(Supplier> 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 51c84f3755..5cb2d12fc5 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 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 @@ */ public class MessagesTcpClient implements MessagesClient { - private final AsyncTcpConnection connection; + private final Supplier connectionSupplier; - public MessagesTcpClient(AsyncTcpConnection connection) { - this.connection = connection; + public MessagesTcpClient(Supplier connectionSupplier) { + this.connectionSupplier = connectionSupplier; + } + + private AsyncTcpConnection connection() { + return connectionSupplier.get(); } @Override @@ -79,7 +84,7 @@ public CompletableFuture pollMessages( 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 CompletableFuture sendMessages( } // 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 c142bf683e..e8544067e3 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.LoggerFactory; import java.util.concurrent.CompletableFuture; +import java.util.function.Supplier; /** * Async TCP implementation of partitions client. @@ -35,10 +36,14 @@ public class PartitionsTcpClient implements PartitionsClient { private static final Logger log = LoggerFactory.getLogger(PartitionsTcpClient.class); - private final AsyncTcpConnection connection; + private final Supplier connectionSupplier; - public PartitionsTcpClient(AsyncTcpConnection connection) { - this.connection = connection; + public PartitionsTcpClient(Supplier connectionSupplier) { + this.connectionSupplier = connectionSupplier; + } + + private AsyncTcpConnection connection() { + return connectionSupplier.get(); } @Override @@ -49,9 +54,11 @@ public CompletableFuture createPartitions(StreamId streamId, TopicId topic 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 CompletableFuture deletePartitions(StreamId streamId, TopicId topic 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 fe31a549de..b5b97842fd 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.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 @@ public class PersonalAccessTokensTcpClient implements PersonalAccessTokensClient { private static final Logger log = LoggerFactory.getLogger(PersonalAccessTokensTcpClient.class); - private final AsyncTcpConnection connection; + private final Supplier connectionSupplier; + private final LoginRedirectionHook redirectionHook; - public PersonalAccessTokensTcpClient(AsyncTcpConnection connection) { - this.connection = connection; + public PersonalAccessTokensTcpClient(Supplier connectionSupplier) { + this(connectionSupplier, LoginRedirectionHook.NONE); + } + + PersonalAccessTokensTcpClient( + Supplier connectionSupplier, LoginRedirectionHook redirectionHook) { + this.connectionSupplier = connectionSupplier; + this.redirectionHook = redirectionHook; + } + + private AsyncTcpConnection connection() { + return connectionSupplier.get(); } @Override @@ -56,7 +68,7 @@ public CompletableFuture createPersonalAccessToken(Strin 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 CompletableFuture> getPersonalAccessTokens( 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 CompletableFuture deletePersonalAccessToken(String name) { 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 CompletableFuture deletePersonalAccessToken(String name) { @Override public CompletableFuture loginWithPersonalAccessToken(String token) { + return loginWithoutRedirect(token).thenCompose(identity -> redirectionHook + .afterLogin(() -> loginWithoutRedirect(token)) + .thenApply(redirectedIdentity -> redirectedIdentity != null ? redirectedIdentity : identity)); + } + + private CompletableFuture 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 0a04668814..91f37e4a54 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.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 @@ */ public class StreamsTcpClient implements StreamsClient { - private final AsyncTcpConnection connection; + private final Supplier connectionSupplier; - public StreamsTcpClient(AsyncTcpConnection connection) { - this.connection = connection; + public StreamsTcpClient(Supplier connectionSupplier) { + this.connectionSupplier = connectionSupplier; + } + + private AsyncTcpConnection connection() { + return connectionSupplier.get(); } @Override @@ -55,7 +60,7 @@ public CompletableFuture createStream(String name) { 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 CompletableFuture createStream(String name) { public CompletableFuture> 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 result; if (response.isReadable()) { result = Optional.of(readStreamDetails(response)); @@ -80,7 +85,7 @@ public CompletableFuture> getStream(StreamId streamId) { @Override public CompletableFuture> getStreams() { - return connection + return connection() .send(CommandCode.Stream.GET_ALL.getValue(), Unpooled.EMPTY_BUFFER) .thenApply(response -> { List streams = new ArrayList<>(); @@ -101,13 +106,13 @@ public CompletableFuture updateStream(StreamId streamId, String name) { 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 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 acf0db698b..8692a2065f 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 @@ 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 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 @@ public class SystemTcpClient implements SystemClient { private static final Logger log = LoggerFactory.getLogger(SystemTcpClient.class); - private final AsyncTcpConnection connection; + private final Supplier connectionSupplier; - public SystemTcpClient(AsyncTcpConnection connection) { - this.connection = connection; + public SystemTcpClient(Supplier connectionSupplier) { + this.connectionSupplier = connectionSupplier; + } + + private AsyncTcpConnection connection() { + return connectionSupplier.get(); } @Override @@ -51,13 +57,32 @@ public CompletableFuture getStats() { 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 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 CompletableFuture getMe() { 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 CompletableFuture getClient(Long clientId) { 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 CompletableFuture> getClients() { 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 CompletableFuture ping() { 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 dce5b70f33..737f0a3410 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.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 @@ */ public class TopicsTcpClient implements TopicsClient { - private final AsyncTcpConnection connection; + private final Supplier connectionSupplier; - public TopicsTcpClient(AsyncTcpConnection connection) { - this.connection = connection; + public TopicsTcpClient(Supplier connectionSupplier) { + this.connectionSupplier = connectionSupplier; + } + + private AsyncTcpConnection connection() { + return connectionSupplier.get(); } @Override @@ -56,7 +61,7 @@ public CompletableFuture> getTopic(StreamId streamId, Top 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 CompletableFuture> getTopic(StreamId streamId, Top public CompletableFuture> 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 topics = new ArrayList<>(); while (response.isReadable()) { @@ -106,7 +111,7 @@ public CompletableFuture createTopic( 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 CompletableFuture updateTopic( 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 CompletableFuture deleteTopic(StreamId streamId, TopicId topicId) { 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 632f45d6d5..6ed7c3bef4 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 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 @@ public class UsersTcpClient implements UsersClient { private static final Logger log = LoggerFactory.getLogger(UsersTcpClient.class); - private final AsyncTcpConnection connection; + private final Supplier connectionSupplier; + private final LoginRedirectionHook redirectionHook; - public UsersTcpClient(AsyncTcpConnection connection) { - this.connection = connection; + public UsersTcpClient(Supplier connectionSupplier) { + this(connectionSupplier, LoginRedirectionHook.NONE); + } + + UsersTcpClient(Supplier connectionSupplier, LoginRedirectionHook redirectionHook) { + this.connectionSupplier = connectionSupplier; + this.redirectionHook = redirectionHook; + } + + private AsyncTcpConnection connection() { + return connectionSupplier.get(); } @Override public CompletableFuture> 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> 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 CompletableFuture createUser( }, () -> payload.writeByte(0)); - return connection.exchangeForEntity(CommandCode.User.CREATE, payload, BytesDeserializer::readUserInfoDetails); + return connection().exchangeForEntity(CommandCode.User.CREATE, payload, BytesDeserializer::readUserInfoDetails); } @Override public CompletableFuture 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 CompletableFuture updateUser(UserId userId, Optional userna }, () -> payload.writeByte(0)); - return connection.sendAndRelease(CommandCode.User.UPDATE, payload); + return connection().sendAndRelease(CommandCode.User.UPDATE, payload); } @Override @@ -120,7 +131,7 @@ public CompletableFuture updatePermissions(UserId userId, Optional 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 CompletableFuture changePassword(UserId userId, String currentPassw 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 login(String username, String password) { + return loginWithoutRedirect(username, password).thenCompose(identity -> redirectionHook + .afterLogin(() -> loginWithoutRedirect(username, password)) + .thenApply(redirectedIdentity -> redirectedIdentity != null ? redirectedIdentity : identity)); + } + + private CompletableFuture loginWithoutRedirect(String username, String password) { String version = IggyVersion.getInstance().getUserAgent(); String context = IggyVersion.getInstance().toString(); @@ -150,9 +167,8 @@ public CompletableFuture login(String username, String password) { 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 CompletableFuture logout() { 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 ecf29e9aa1..abb7b42601 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 3507919dcb..d238c32f01 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 @@ 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 @@ public Stats getStats() { 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 2bf5eecde5..8a60c27bcc 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 IdentityInfo login() { 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 1d1df7fa48..4d00cfa147 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 @@ public Stats getStats() { 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/cluster/ClusterMetadata.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/cluster/ClusterMetadata.java new file mode 100644 index 0000000000..757108c615 --- /dev/null +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/cluster/ClusterMetadata.java @@ -0,0 +1,33 @@ +/* + * 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 java.util.List; + +/** + * Cluster metadata reported by the server. + * + *

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 nodes) {} diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/cluster/ClusterNode.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/cluster/ClusterNode.java new file mode 100644 index 0000000000..1114fca880 --- /dev/null +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/cluster/ClusterNode.java @@ -0,0 +1,32 @@ +/* + * 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; + +/** + * 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/cluster/ClusterNodeRole.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/cluster/ClusterNodeRole.java new file mode 100644 index 0000000000..d95f7532a0 --- /dev/null +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/cluster/ClusterNodeRole.java @@ -0,0 +1,52 @@ +/* + * 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; + +/** + * 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 0000000000..be115c5b63 --- /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/cluster/TransportEndpoints.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/cluster/TransportEndpoints.java new file mode 100644 index 0000000000..4a4a68ec40 --- /dev/null +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/cluster/TransportEndpoints.java @@ -0,0 +1,31 @@ +/* + * 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; + +/** + * 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/cluster/package-info.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/cluster/package-info.java new file mode 100644 index 0000000000..c3385a3f6a --- /dev/null +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/cluster/package-info.java @@ -0,0 +1,23 @@ +/* + * 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. + */ + +@NonNullApi +package org.apache.iggy.cluster; + +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 7b4328b0cc..67afa08523 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 @@ 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 @@ */ 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 static ConsumerGroupInfo readConsumerGroupInfo(ByteBuf response) { 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 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 static PersonalAccessTokenInfo readPersonalAccessTokenInfo(ByteBuf respon 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 2363947070..337494d185 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 0000000000..fe542542fd --- /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 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 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 0000000000..00540cd23c --- /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 findLeader(Supplier> 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 01d55315b1..550b69b40b 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 @@ void shouldGetStats() { 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 db00a876fb..6337a502b9 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 @@ void shouldGetClients() { 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 f592ce5a5b..637456ab1d 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 @@ 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 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.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 @@ void shouldDeserializePersonalAccessTokenInfoWithoutExpiry() { 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 cb86d7b8b4..cfc0007969 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})"