Skip to content

Commit ee2f87f

Browse files
ggivoggivo
authored
[automatic-failover] Enforce formatting for automatic failover codebase (#4176)
* enforce formating for mcf * enforce formating for mcf * Trigger integration test jobs feature branches
1 parent de0e4be commit ee2f87f

File tree

9 files changed

+162
-136
lines changed

9 files changed

+162
-136
lines changed

‎.github/workflows/test-on-docker.yml‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
---
22

3-
name: Build and Test using containerized environment
3+
name: Build and Test using a containerized environment
44

55
on:
66
push:
@@ -11,10 +11,12 @@ on:
1111
branches:
1212
- master
1313
- '[0-9].*'
14+
- 'feature/**'
1415
pull_request:
1516
branches:
1617
- master
1718
- '[0-9].*'
19+
- 'feature/**'
1820
schedule:
1921
- cron: '0 1 * * *' # nightly build
2022
workflow_dispatch:

‎pom.xml‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -334,6 +334,8 @@
334334
<configFile>${project.basedir}/hbase-formatter.xml</configFile>
335335
<directories>
336336
<directory>${project.basedir}/src/main/java/redis/clients/jedis/annots</directory>
337+
<directory>${project.basedir}/src/main/java/redis/clients/jedis/mcf</directory>
338+
<directory>${project.basedir}/src/test/java/redis/clients/jedis/failover</directory>
337339
</directories>
338340
</configuration>
339341
<executions>

‎src/main/java/redis/clients/jedis/mcf/CircuitBreakerCommandExecutor.java‎

Lines changed: 40 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -13,51 +13,57 @@
1313

1414
/**
1515
* @author Allen Terleto (aterleto)
16-
* <p>
17-
* CommandExecutor with built-in retry, circuit-breaker, and failover to another cluster/database endpoint.
18-
* With this executor users can seamlessly failover to Disaster Recovery (DR), Backup, and Active-Active cluster(s)
19-
* by using simple configuration which is passed through from Resilience4j - https://resilience4j.readme.io/docs
20-
* <p>
16+
* <p>
17+
* CommandExecutor with built-in retry, circuit-breaker, and failover to another
18+
* cluster/database endpoint. With this executor users can seamlessly failover to Disaster
19+
* Recovery (DR), Backup, and Active-Active cluster(s) by using simple configuration which
20+
* is passed through from Resilience4j - https://resilience4j.readme.io/docs
21+
* <p>
2122
*/
2223
@Experimental
23-
public class CircuitBreakerCommandExecutor extends CircuitBreakerFailoverBase implements CommandExecutor {
24+
public class CircuitBreakerCommandExecutor extends CircuitBreakerFailoverBase
25+
implements CommandExecutor {
2426

25-
public CircuitBreakerCommandExecutor(MultiClusterPooledConnectionProvider provider) {
26-
super(provider);
27-
}
27+
public CircuitBreakerCommandExecutor(MultiClusterPooledConnectionProvider provider) {
28+
super(provider);
29+
}
2830

29-
@Override
30-
public <T> T executeCommand(CommandObject<T> commandObject) {
31-
Cluster cluster = provider.getCluster(); // Pass this by reference for thread safety
31+
@Override
32+
public <T> T executeCommand(CommandObject<T> commandObject) {
33+
Cluster cluster = provider.getCluster(); // Pass this by reference for thread safety
3234

33-
DecorateSupplier<T> supplier = Decorators.ofSupplier(() -> this.handleExecuteCommand(commandObject, cluster));
35+
DecorateSupplier<T> supplier = Decorators
36+
.ofSupplier(() -> this.handleExecuteCommand(commandObject, cluster));
3437

35-
supplier.withRetry(cluster.getRetry());
36-
supplier.withCircuitBreaker(cluster.getCircuitBreaker());
37-
supplier.withFallback(provider.getFallbackExceptionList(),
38-
e -> this.handleClusterFailover(commandObject, cluster.getCircuitBreaker()));
38+
supplier.withRetry(cluster.getRetry());
39+
supplier.withCircuitBreaker(cluster.getCircuitBreaker());
40+
supplier.withFallback(provider.getFallbackExceptionList(),
41+
e -> this.handleClusterFailover(commandObject, cluster.getCircuitBreaker()));
3942

40-
return supplier.decorate().get();
41-
}
43+
return supplier.decorate().get();
44+
}
4245

43-
/**
44-
* Functional interface wrapped in retry and circuit breaker logic to handle happy path scenarios
45-
*/
46-
private <T> T handleExecuteCommand(CommandObject<T> commandObject, Cluster cluster) {
47-
try (Connection connection = cluster.getConnection()) {
48-
return connection.executeCommand(commandObject);
49-
}
46+
/**
47+
* Functional interface wrapped in retry and circuit breaker logic to handle happy path scenarios
48+
*/
49+
private <T> T handleExecuteCommand(CommandObject<T> commandObject, Cluster cluster) {
50+
try (Connection connection = cluster.getConnection()) {
51+
return connection.executeCommand(commandObject);
5052
}
53+
}
5154

52-
/**
53-
* Functional interface wrapped in retry and circuit breaker logic to handle open circuit breaker failure scenarios
54-
*/
55-
private <T> T handleClusterFailover(CommandObject<T> commandObject, CircuitBreaker circuitBreaker) {
55+
/**
56+
* Functional interface wrapped in retry and circuit breaker logic to handle open circuit breaker
57+
* failure scenarios
58+
*/
59+
private <T> T handleClusterFailover(CommandObject<T> commandObject,
60+
CircuitBreaker circuitBreaker) {
5661

57-
clusterFailover(circuitBreaker);
62+
clusterFailover(circuitBreaker);
5863

59-
// Recursive call to the initiating method so the operation can be retried on the next cluster connection
60-
return executeCommand(commandObject);
61-
}
64+
// Recursive call to the initiating method so the operation can be retried on the next cluster
65+
// connection
66+
return executeCommand(commandObject);
67+
}
6268

6369
}

‎src/main/java/redis/clients/jedis/mcf/CircuitBreakerFailoverBase.java‎

Lines changed: 51 additions & 42 deletions
Original file line numberDiff line numberDiff line change
@@ -10,59 +10,68 @@
1010

1111
/**
1212
* @author Allen Terleto (aterleto)
13-
* <p>
14-
* Base class for CommandExecutor with built-in retry, circuit-breaker, and failover to another cluster/database
15-
* endpoint. With this executor users can seamlessly failover to Disaster Recovery (DR), Backup, and Active-Active
16-
* cluster(s) by using simple configuration which is passed through from
17-
* Resilience4j - https://resilience4j.readme.io/docs
18-
* <p>
13+
* <p>
14+
* Base class for CommandExecutor with built-in retry, circuit-breaker, and failover to
15+
* another cluster/database endpoint. With this executor users can seamlessly failover to
16+
* Disaster Recovery (DR), Backup, and Active-Active cluster(s) by using simple
17+
* configuration which is passed through from Resilience4j -
18+
* https://resilience4j.readme.io/docs
19+
* <p>
1920
*/
2021
@Experimental
2122
public class CircuitBreakerFailoverBase implements AutoCloseable {
22-
private final Lock lock = new ReentrantLock(true);
23+
private final Lock lock = new ReentrantLock(true);
2324

24-
protected final MultiClusterPooledConnectionProvider provider;
25+
protected final MultiClusterPooledConnectionProvider provider;
2526

26-
public CircuitBreakerFailoverBase(MultiClusterPooledConnectionProvider provider) {
27-
this.provider = provider;
28-
}
27+
public CircuitBreakerFailoverBase(MultiClusterPooledConnectionProvider provider) {
28+
this.provider = provider;
29+
}
2930

30-
@Override
31-
public void close() {
32-
IOUtils.closeQuietly(this.provider);
33-
}
31+
@Override
32+
public void close() {
33+
IOUtils.closeQuietly(this.provider);
34+
}
35+
36+
/**
37+
* Functional interface wrapped in retry and circuit breaker logic to handle open circuit breaker
38+
* failure scenarios
39+
*/
40+
protected void clusterFailover(CircuitBreaker circuitBreaker) {
41+
lock.lock();
3442

35-
/**
36-
* Functional interface wrapped in retry and circuit breaker logic to handle open circuit breaker failure scenarios
37-
*/
38-
protected void clusterFailover(CircuitBreaker circuitBreaker) {
39-
lock.lock();
40-
41-
try {
42-
// Check state to handle race conditions since incrementActiveMultiClusterIndex() is non-idempotent
43-
if (!CircuitBreaker.State.FORCED_OPEN.equals(circuitBreaker.getState())) {
43+
try {
44+
// Check state to handle race conditions since incrementActiveMultiClusterIndex() is
45+
// non-idempotent
46+
if (!CircuitBreaker.State.FORCED_OPEN.equals(circuitBreaker.getState())) {
4447

45-
// Transitions state machine to a FORCED_OPEN state, stopping state transition, metrics and event publishing.
46-
// To recover/transition from this forced state the user will need to manually failback
47-
circuitBreaker.transitionToForcedOpenState();
48+
// Transitions state machine to a FORCED_OPEN state, stopping state transition, metrics and
49+
// event publishing.
50+
// To recover/transition from this forced state the user will need to manually failback
51+
circuitBreaker.transitionToForcedOpenState();
4852

49-
// Incrementing the activeMultiClusterIndex will allow subsequent calls to the executeCommand()
50-
// to use the next cluster's connection pool - according to the configuration's prioritization/order
51-
int activeMultiClusterIndex = provider.incrementActiveMultiClusterIndex();
53+
// Incrementing the activeMultiClusterIndex will allow subsequent calls to the
54+
// executeCommand()
55+
// to use the next cluster's connection pool - according to the configuration's
56+
// prioritization/order
57+
int activeMultiClusterIndex = provider.incrementActiveMultiClusterIndex();
5258

53-
// Implementation is optionally provided during configuration. Typically, used for activeMultiClusterIndex persistence or custom logging
54-
provider.runClusterFailoverPostProcessor(activeMultiClusterIndex);
55-
}
59+
// Implementation is optionally provided during configuration. Typically, used for
60+
// activeMultiClusterIndex persistence or custom logging
61+
provider.runClusterFailoverPostProcessor(activeMultiClusterIndex);
62+
}
5663

57-
// Once the priority list is exhausted only a manual failback can open the circuit breaker so all subsequent operations will fail
58-
else if (provider.isLastClusterCircuitBreakerForcedOpen()) {
59-
throw new JedisConnectionException("Cluster/database endpoint could not failover since the MultiClusterClientConfig was not " +
60-
"provided with an additional cluster/database endpoint according to its prioritized sequence. " +
61-
"If applicable, consider failing back OR restarting with an available cluster/database endpoint");
62-
}
63-
} finally {
64-
lock.unlock();
65-
}
64+
// Once the priority list is exhausted only a manual failback can open the circuit breaker so
65+
// all subsequent operations will fail
66+
else if (provider.isLastClusterCircuitBreakerForcedOpen()) {
67+
throw new JedisConnectionException(
68+
"Cluster/database endpoint could not failover since the MultiClusterClientConfig was not "
69+
+ "provided with an additional cluster/database endpoint according to its prioritized sequence. "
70+
+ "If applicable, consider failing back OR restarting with an available cluster/database endpoint");
71+
}
72+
} finally {
73+
lock.unlock();
6674
}
75+
}
6776

6877
}

‎src/main/java/redis/clients/jedis/mcf/CircuitBreakerFailoverConnectionProvider.java‎

Lines changed: 43 additions & 39 deletions
Original file line numberDiff line numberDiff line change
@@ -10,48 +10,52 @@
1010
import redis.clients.jedis.providers.MultiClusterPooledConnectionProvider.Cluster;
1111

1212
/**
13-
* ConnectionProvider with built-in retry, circuit-breaker, and failover to another cluster/database endpoint.
14-
* With this executor users can seamlessly failover to Disaster Recovery (DR), Backup, and Active-Active cluster(s)
15-
* by using simple configuration which is passed through from Resilience4j - https://resilience4j.readme.io/docs
13+
* ConnectionProvider with built-in retry, circuit-breaker, and failover to another cluster/database
14+
* endpoint. With this executor users can seamlessly failover to Disaster Recovery (DR), Backup, and
15+
* Active-Active cluster(s) by using simple configuration which is passed through from Resilience4j
16+
* - https://resilience4j.readme.io/docs
1617
*/
1718
@Experimental
1819
public class CircuitBreakerFailoverConnectionProvider extends CircuitBreakerFailoverBase {
1920

20-
public CircuitBreakerFailoverConnectionProvider(MultiClusterPooledConnectionProvider provider) {
21-
super(provider);
22-
}
23-
24-
public Connection getConnection() {
25-
Cluster cluster = provider.getCluster(); // Pass this by reference for thread safety
26-
27-
DecorateSupplier<Connection> supplier = Decorators.ofSupplier(() -> this.handleGetConnection(cluster));
28-
29-
supplier.withRetry(cluster.getRetry());
30-
supplier.withCircuitBreaker(cluster.getCircuitBreaker());
31-
supplier.withFallback(provider.getFallbackExceptionList(),
32-
e -> this.handleClusterFailover(cluster.getCircuitBreaker()));
33-
34-
return supplier.decorate().get();
35-
}
36-
37-
/**
38-
* Functional interface wrapped in retry and circuit breaker logic to handle happy path scenarios
39-
*/
40-
private Connection handleGetConnection(Cluster cluster) {
41-
Connection connection = cluster.getConnection();
42-
connection.ping();
43-
return connection;
44-
}
45-
46-
/**
47-
* Functional interface wrapped in retry and circuit breaker logic to handle open circuit breaker failure scenarios
48-
*/
49-
private Connection handleClusterFailover(CircuitBreaker circuitBreaker) {
50-
51-
clusterFailover(circuitBreaker);
52-
53-
// Recursive call to the initiating method so the operation can be retried on the next cluster connection
54-
return getConnection();
55-
}
21+
public CircuitBreakerFailoverConnectionProvider(MultiClusterPooledConnectionProvider provider) {
22+
super(provider);
23+
}
24+
25+
public Connection getConnection() {
26+
Cluster cluster = provider.getCluster(); // Pass this by reference for thread safety
27+
28+
DecorateSupplier<Connection> supplier = Decorators
29+
.ofSupplier(() -> this.handleGetConnection(cluster));
30+
31+
supplier.withRetry(cluster.getRetry());
32+
supplier.withCircuitBreaker(cluster.getCircuitBreaker());
33+
supplier.withFallback(provider.getFallbackExceptionList(),
34+
e -> this.handleClusterFailover(cluster.getCircuitBreaker()));
35+
36+
return supplier.decorate().get();
37+
}
38+
39+
/**
40+
* Functional interface wrapped in retry and circuit breaker logic to handle happy path scenarios
41+
*/
42+
private Connection handleGetConnection(Cluster cluster) {
43+
Connection connection = cluster.getConnection();
44+
connection.ping();
45+
return connection;
46+
}
47+
48+
/**
49+
* Functional interface wrapped in retry and circuit breaker logic to handle open circuit breaker
50+
* failure scenarios
51+
*/
52+
private Connection handleClusterFailover(CircuitBreaker circuitBreaker) {
53+
54+
clusterFailover(circuitBreaker);
55+
56+
// Recursive call to the initiating method so the operation can be retried on the next cluster
57+
// connection
58+
return getConnection();
59+
}
5660

5761
}

‎src/main/java/redis/clients/jedis/mcf/MultiClusterPipeline.java‎

Lines changed: 7 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,8 @@
1212

1313
/**
1414
* This is high memory dependent solution as all the appending commands will be hold in memory until
15-
* {@link MultiClusterPipeline#sync() SYNC} (or {@link MultiClusterPipeline#close() CLOSE}) gets called.
15+
* {@link MultiClusterPipeline#sync() SYNC} (or {@link MultiClusterPipeline#close() CLOSE}) gets
16+
* called.
1617
*/
1718
@Experimental
1819
public class MultiClusterPipeline extends PipelineBase implements Closeable {
@@ -32,7 +33,8 @@ public MultiClusterPipeline(MultiClusterPooledConnectionProvider pooledProvider)
3233
}
3334
}
3435

35-
public MultiClusterPipeline(MultiClusterPooledConnectionProvider pooledProvider, CommandObjects commandObjects) {
36+
public MultiClusterPipeline(MultiClusterPooledConnectionProvider pooledProvider,
37+
CommandObjects commandObjects) {
3638
super(commandObjects);
3739
this.failoverProvider = new CircuitBreakerFailoverConnectionProvider(pooledProvider);
3840
}
@@ -52,8 +54,9 @@ public void close() {
5254
}
5355

5456
/**
55-
* Synchronize pipeline by reading all responses. This operation close the pipeline. In order to get return values
56-
* from pipelined commands, capture the different Response&lt;?&gt; of the commands you execute.
57+
* Synchronize pipeline by reading all responses. This operation close the pipeline. In order to
58+
* get return values from pipelined commands, capture the different Response&lt;?&gt; of the
59+
* commands you execute.
5760
*/
5861
@Override
5962
public void sync() {

‎src/main/java/redis/clients/jedis/mcf/MultiClusterTransaction.java‎

Lines changed: 4 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,7 @@
2424
public class MultiClusterTransaction extends TransactionBase {
2525

2626
private static final Builder<?> NO_OP_BUILDER = BuilderFactory.RAW_OBJECT;
27-
27+
2828
private static final String GRAPH_COMMANDS_NOT_SUPPORTED_MESSAGE = "Graph commands are not supported.";
2929

3030
private final CircuitBreakerFailoverConnectionProvider failoverProvider;
@@ -47,7 +47,6 @@ public MultiClusterTransaction(MultiClusterPooledConnectionProvider provider) {
4747
/**
4848
* A user wanting to WATCH/UNWATCH keys followed by a call to MULTI ({@link #multi()}) it should
4949
* be {@code doMulti=false}.
50-
*
5150
* @param provider
5251
* @param doMulti {@code false} should be set to enable manual WATCH, UNWATCH and MULTI
5352
*/
@@ -66,12 +65,12 @@ public MultiClusterTransaction(MultiClusterPooledConnectionProvider provider, bo
6665
/**
6766
* A user wanting to WATCH/UNWATCH keys followed by a call to MULTI ({@link #multi()}) it should
6867
* be {@code doMulti=false}.
69-
*
7068
* @param provider
7169
* @param doMulti {@code false} should be set to enable manual WATCH, UNWATCH and MULTI
7270
* @param commandObjects command objects
7371
*/
74-
public MultiClusterTransaction(MultiClusterPooledConnectionProvider provider, boolean doMulti, CommandObjects commandObjects) {
72+
public MultiClusterTransaction(MultiClusterPooledConnectionProvider provider, boolean doMulti,
73+
CommandObjects commandObjects) {
7574
super(commandObjects);
7675
this.failoverProvider = new CircuitBreakerFailoverConnectionProvider(provider);
7776

@@ -169,7 +168,7 @@ public final List<Object> exec() {
169168
}
170169

171170
List<Object> formatted = new ArrayList<>(unformatted.size() - extraCommandCount.get());
172-
for (Object rawReply: unformatted) {
171+
for (Object rawReply : unformatted) {
173172
try {
174173
Response<?> response = commands.poll().getValue();
175174
response.set(rawReply);

0 commit comments

Comments
 (0)