Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions src/main/java/io/tarantool/driver/api/TarantoolClient.java
Original file line number Diff line number Diff line change
Expand Up @@ -26,19 +26,22 @@ public interface TarantoolClient<T extends Packable, R extends Collection<T>>

/**
* Get the Tarantool client config passed to this client
*
* @return {@link TarantoolClientConfig} instance
*/
TarantoolClientConfig getConfig();

/**
* Get the Tarantool server version
*
* @return {@link TarantoolVersion}
* @throws TarantoolClientException if the client is not connected
*/
TarantoolVersion getVersion() throws TarantoolClientException;

/**
* Provides CRUD and other operations for a Tarantool space
*
* @param spaceName name of the space, must not be null or empty
* @return Tarantool space operations interface
* @throws TarantoolClientException if the client is not connected
Expand All @@ -47,6 +50,7 @@ public interface TarantoolClient<T extends Packable, R extends Collection<T>>

/**
* Provides CRUD and other operations for a Tarantool space
*
* @param spaceId ID of the space, must be greater than 0
* @return Tarantool space operations implementation
* @throws TarantoolClientException if the client is not connected
Expand All @@ -55,6 +59,7 @@ public interface TarantoolClient<T extends Packable, R extends Collection<T>>

/**
* Provides operations for Tarantool spaces and indexes metadata
*
* @return Tarantool metadata operations implementation
* @throws TarantoolClientException if the client is not connected
*/
Expand All @@ -67,4 +72,11 @@ public interface TarantoolClient<T extends Packable, R extends Collection<T>>
* @return connection listeners
*/
TarantoolConnectionListeners getConnectionListeners();

/**
* Starts the process of establishing lacking connections to each host
*
* @return returns true if the establishing process has been started, else false
*/
boolean establishLackingConnections();
}
Original file line number Diff line number Diff line change
Expand Up @@ -148,6 +148,11 @@ private TarantoolConnectionManager connectionManager() {
return this.connectionManager;
}

@Override
public boolean establishLackingConnections() {
return connectionManager.establishLackingConnections();
}

@Override
public TarantoolVersion getVersion() throws TarantoolClientException {
try {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -416,6 +416,11 @@ public CompletableFuture<List<?>> eval(String expression,
return client.eval(expression, arguments, argumentsMapper, resultMapper);
}

@Override
public boolean establishLackingConnections() {
return this.client.establishLackingConnections();
}

@Override
public void close() throws Exception {
this.client.close();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -378,6 +378,11 @@ public CompletableFuture<List<?>> eval(String expression, List<?> arguments,
return wrapOperation(() -> client.eval(expression, arguments, argumentsMapper, resultMapper));
}

@Override
public boolean establishLackingConnections() {
return this.client.establishLackingConnections();
}

@Override
public void close() throws Exception {
client.close();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -104,6 +104,11 @@ public CompletableFuture<TarantoolConnection> getConnection() {
});
}

@Override
public boolean establishLackingConnections() {
return connectionMode.compareAndSet(ConnectionMode.OFF, ConnectionMode.PARTIAL);
}

private CompletableFuture<TarantoolConnection> getConnectionInternal() {
CompletableFuture<TarantoolConnection> result;
ConnectionMode currentMode = connectionMode.get();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -18,4 +18,11 @@ public interface TarantoolConnectionManager extends AutoCloseable {
* @return a future with next connection in order
*/
CompletableFuture<TarantoolConnection> getConnection();

/**
* Starts the process of establishing lacking connections to each host
*
* @return returns true if the establishing process has been started, else false
*/
boolean establishLackingConnections();
}
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@ public void httpClusterDiscovererTest() throws TarantoolClientException {
HTTPDiscoveryClusterAddressProvider addressProvider = getHttpProvider();
Collection<TarantoolServerAddress> nodes = addressProvider.getAddresses();

assertEquals(nodes.size(), 2);
assertEquals(nodes.size(), 3);
Set<TarantoolServerAddress> nodeSet = new HashSet<>(nodes);
assertTrue(nodeSet.contains(new TarantoolServerAddress(TEST_ROUTER1_URI)));
assertTrue(nodeSet.contains(new TarantoolServerAddress(TEST_ROUTER2_URI)));
Expand All @@ -68,7 +68,7 @@ public void binaryClusterDiscovererTest() {
TarantoolClusterAddressProvider addressProvider = getBinaryProvider();

Collection<TarantoolServerAddress> nodes = addressProvider.getAddresses();
assertEquals(nodes.size(), 2);
assertEquals(nodes.size(), 3);
Set<TarantoolServerAddress> nodeSet = new HashSet<>(nodes);
assertTrue(nodeSet.contains(new TarantoolServerAddress(TEST_ROUTER1_URI)));
assertTrue(nodeSet.contains(new TarantoolServerAddress(TEST_ROUTER2_URI)));
Expand Down
98 changes: 98 additions & 0 deletions src/test/java/io/tarantool/driver/integration/ReconnectIT.java
Original file line number Diff line number Diff line change
@@ -0,0 +1,98 @@
package io.tarantool.driver.integration;

import io.tarantool.driver.api.TarantoolClient;
import io.tarantool.driver.api.TarantoolClientFactory;
import io.tarantool.driver.api.TarantoolResult;
import io.tarantool.driver.api.TarantoolServerAddress;
import io.tarantool.driver.api.tuple.TarantoolTuple;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.testcontainers.containers.output.WaitingConsumer;
import org.testcontainers.junit.jupiter.Testcontainers;

import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.TimeoutException;

import static org.junit.jupiter.api.Assertions.assertEquals;

@Testcontainers
public class ReconnectIT extends SharedCartridgeContainer {

private static String USER_NAME;
private static String PASSWORD;

@BeforeAll
public static void setUp() throws TimeoutException {
startCluster();

WaitingConsumer waitingConsumer = new WaitingConsumer();
container.followOutput(waitingConsumer);
waitingConsumer.waitUntil(f -> f.getUtf8String().contains("The cluster is balanced ok"));

USER_NAME = container.getUsername();
PASSWORD = container.getPassword();
}

@Test
public void test_should_reconnect_ifReconnectIsInvoked() throws Exception {
//when
TarantoolClient<TarantoolTuple, TarantoolResult<TarantoolTuple>> client =
TarantoolClientFactory.createClient()
.withAddresses(
new TarantoolServerAddress(container.getRouterHost(), container.getMappedPort(3301)),
new TarantoolServerAddress(container.getRouterHost(), container.getMappedPort(3311)),
new TarantoolServerAddress(container.getRouterHost(), container.getMappedPort(3312))
)
.withCredentials(USER_NAME, PASSWORD)
.withConnections(10)
.build();

// getting all routers uuids
final Set<String> routerUuids = getInstancesUuids(client);

// stop routers
container.execInContainer("cartridge", "stop", "--run-dir=/tmp/run", "router");
container.execInContainer("cartridge", "stop", "--run-dir=/tmp/run", "second-router");

// check that there is only one instance left
assertEquals(getInstanceUuid(client), getInstanceUuid(client));

// start routers
container.execInContainer("cartridge", "start", "--run-dir=/tmp/run", "--data-dir=/tmp/data", "-d");

client.establishLackingConnections();
Thread.sleep(3000);

// getting all routers uuids after restarting
final Set<String> uuidsAfterReconnect = getInstancesUuids(client);

// check that amount of routers is equal initial amount
assertEquals(routerUuids.size(), uuidsAfterReconnect.size());
}

/**
* Return all instances uuids from cluster, using round robin connection selection strategy
*
* @param client Tarantool client
* @return set of instances uuids from cluster
*/
private Set<String> getInstancesUuids(TarantoolClient<TarantoolTuple, TarantoolResult<TarantoolTuple>> client) {
String firstUuid = getInstanceUuid(client);

final Set<String> routerUuids = new HashSet<>();
routerUuids.add(firstUuid);

String currentUuid = "";
while (!firstUuid.equals(currentUuid)) {
currentUuid = getInstanceUuid(client);
routerUuids.add(currentUuid);
}

return routerUuids;
}

private String getInstanceUuid(TarantoolClient<TarantoolTuple, TarantoolResult<TarantoolTuple>> client) {
return (String) client.eval("return box.info().uuid").join().get(0);
}
}
5 changes: 5 additions & 0 deletions src/test/resources/cartridge/instances.yml
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,11 @@ testapp.second-router:
advertise_uri: localhost:3311
http_port: 8091

testapp.third-router:
workdir: ./tmp/db_dev/3312
advertise_uri: localhost:3312
http_port: 8092

testapp.s1-master:
workdir: ./tmp/db_dev/3302
advertise_uri: localhost:3302
Expand Down
40 changes: 22 additions & 18 deletions src/test/resources/cartridge/topology.lua
Original file line number Diff line number Diff line change
@@ -1,19 +1,23 @@
cartridge = require('cartridge')
replicasets = {{
alias = 'app-router',
roles = {'vshard-router', 'app.roles.custom', 'app.roles.api_router'},
join_servers = {{uri = 'localhost:3301'}}
}, {
alias = 'app-router-second',
roles = {'vshard-router', 'app.roles.custom', 'app.roles.api_router'},
join_servers = {{uri = 'localhost:3311'}}
}, {
alias = 's1-storage',
roles = {'vshard-storage', 'app.roles.api_storage'},
join_servers = {{uri = 'localhost:3302'}, {uri = 'localhost:3303'}}
}, {
alias = 's2-storage',
roles = {'vshard-storage', 'app.roles.api_storage'},
join_servers = {{uri = 'localhost:3304'}, {uri = 'localhost:3305'}}
}}
return cartridge.admin_edit_topology({replicasets = replicasets})
replicasets = { {
alias = 'app-router',
roles = { 'vshard-router', 'app.roles.custom', 'app.roles.api_router' },
join_servers = { { uri = 'localhost:3301' } }
}, {
alias = 'app-router-second',
roles = { 'vshard-router', 'app.roles.custom', 'app.roles.api_router' },
join_servers = { { uri = 'localhost:3311' } }
}, {
alias = 'app-router-third',
roles = { 'vshard-router', 'app.roles.custom', 'app.roles.api_router' },
join_servers = { { uri = 'localhost:3312' } }
}, {
alias = 's1-storage',
roles = { 'vshard-storage', 'app.roles.api_storage' },
join_servers = { { uri = 'localhost:3302' }, { uri = 'localhost:3303' } }
}, {
alias = 's2-storage',
roles = { 'vshard-storage', 'app.roles.api_storage' },
join_servers = { { uri = 'localhost:3304' }, { uri = 'localhost:3305' } }
} }
return cartridge.admin_edit_topology({ replicasets = replicasets })