Failover handling improvements

pull/1705/head
Nikita 7 years ago
parent 9d508ac3a3
commit d160fddd15

@ -506,11 +506,12 @@ public class MasterSlaveConnectionManager implements ConnectionManager {
protected final void changeMaster(int slot, URI address) {
final MasterSlaveEntry entry = getEntry(slot);
client2entry.remove(entry.getClient());
final RedisClient oldClient = entry.getClient();
entry.changeMaster(address).addListener(new FutureListener<RedisClient>() {
@Override
public void operationComplete(Future<RedisClient> future) throws Exception {
if (future.isSuccess()) {
client2entry.remove(oldClient);
client2entry.put(entry.getClient(), entry);
}
}

@ -127,6 +127,7 @@ public class MasterSlaveEntry {
@Override
public void operationComplete(Future<InetSocketAddress> future) throws Exception {
if (!future.isSuccess()) {
client.shutdownAsync();
result.tryFailure(future.cause());
return;
}
@ -304,6 +305,14 @@ public class MasterSlaveEntry {
}
}
RFuture<Void> addFuture = slaveBalancer.add(entry);
addFuture.addListener(new FutureListener<Void>() {
@Override
public void operationComplete(Future<Void> future) throws Exception {
if (!future.isSuccess()) {
client.shutdownAsync();
}
}
});
addFuture.addListener(new TransferListener<Void>(result));
}
});

@ -73,8 +73,10 @@ abstract class ConnectionPool<T extends RedisConnection> {
promise.addListener(new FutureListener<Void>() {
@Override
public void operationComplete(Future<Void> future) throws Exception {
if (future.isSuccess()) {
entries.add(entry);
}
}
});
initConnections(entry, promise, true);
return promise;
@ -119,13 +121,34 @@ abstract class ConnectionPool<T extends RedisConnection> {
public void operationComplete(Future<T> future) throws Exception {
if (future.isSuccess()) {
T conn = future.getNow();
if (!initPromise.isDone()) {
releaseConnection(entry, conn);
} else {
conn.closeAsync();
}
}
releaseConnection(entry);
if (!future.isSuccess()) {
if (initPromise.isDone()) {
return;
}
for (RedisConnection connection : entry.getAllConnections()) {
if (!connection.isClosed()) {
connection.closeAsync();
}
}
entry.getAllConnections().clear();
for (RedisConnection connection : entry.getAllSubscribeConnections()) {
if (!connection.isClosed()) {
connection.closeAsync();
}
}
entry.getAllSubscribeConnections().clear();
int totalInitializedConnections = minimumIdleSize - initializedConnections.get();
String errorMsg;
if (totalInitializedConnections == 0) {
@ -141,9 +164,8 @@ abstract class ConnectionPool<T extends RedisConnection> {
int value = initializedConnections.decrementAndGet();
if (value == 0) {
if (initPromise.trySuccess(null)) {
log.info("{} connections initialized for {}", minimumIdleSize, entry.getClient().getAddr());
if (!initPromise.trySuccess(null)) {
throw new IllegalStateException();
}
} else if (value > 0 && !initPromise.isDone()) {
if (requests.incrementAndGet() <= minimumIdleSize) {

Loading…
Cancel
Save