MirrorMirror
/
Signal-Server
mirror of https://github.com/signalapp/Signal-Server.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Don't retry pub/sub commands

This commit is contained in:
Jon Chambers 2024-11-01 11:18:50 -04:00 committed by Jon Chambers
parent c9a396b9e3
commit 00d0dba62c
6 changed files with 10 additions and 46 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
service/src/main/java/org/whispersystems/textsecuregcm/redis/AbstractFaultTolerantPubSubConnection.java
View File

@ -7,7 +7,6 @@ package org.whispersystems.textsecuregcm.redis;
import static org.whispersystems.textsecuregcm.metrics.MetricsUtil.name;
import io.github.resilience4j.retry.Retry;
import io.lettuce.core.RedisException;
import io.lettuce.core.pubsub.StatefulRedisPubSubConnection;
import io.micrometer.core.instrument.Metrics;
@ -20,17 +19,11 @@ abstract class AbstractFaultTolerantPubSubConnection<K, V, C extends StatefulRed
private final String name;
private final C pubSubConnection;
private final Retry retry;
private final Timer executeTimer;
protected AbstractFaultTolerantPubSubConnection(final String name,
final C pubSubConnection,
final Retry retry) {
protected AbstractFaultTolerantPubSubConnection(final String name, final C pubSubConnection) {
this.name = name;
this.pubSubConnection = pubSubConnection;
this.retry = retry;
this.executeTimer = Metrics.timer(name(getClass(), "execute"), "clusterName", name + "-pubsub");
}
@ -41,7 +34,7 @@ abstract class AbstractFaultTolerantPubSubConnection<K, V, C extends StatefulRed
public void usePubSubConnection(final Consumer<C> consumer) {
try {
retry.executeRunnable(() -> executeTimer.record(() -> consumer.accept(pubSubConnection)));
executeTimer.record(() -> consumer.accept(pubSubConnection));
} catch (final Throwable t) {
if (t instanceof RedisException) {
throw (RedisException) t;
@ -53,7 +46,7 @@ abstract class AbstractFaultTolerantPubSubConnection<K, V, C extends StatefulRed
public <T> T withPubSubConnection(final Function<C, T> function) {
try {
return retry.executeCallable(() -> executeTimer.record(() -> function.apply(pubSubConnection)));
return executeTimer.record(() -> function.apply(pubSubConnection));
} catch (final Throwable t) {
if (t instanceof RedisException) {
throw (RedisException) t;

3
service/src/main/java/org/whispersystems/textsecuregcm/redis/FaultTolerantPubSubClusterConnection.java
View File

@ -21,11 +21,10 @@ public class FaultTolerantPubSubClusterConnection<K, V> extends AbstractFaultTol
protected FaultTolerantPubSubClusterConnection(final String name,
final StatefulRedisClusterPubSubConnection<K, V> pubSubConnection,
final Retry retry,
final Retry resubscribeRetry,
final Scheduler topologyChangedEventScheduler) {
super(name, pubSubConnection, retry);
super(name, pubSubConnection);
pubSubConnection.setNodeMessagePropagation(true);

6
service/src/main/java/org/whispersystems/textsecuregcm/redis/FaultTolerantPubSubConnection.java
View File

@ -5,15 +5,13 @@
package org.whispersystems.textsecuregcm.redis;
import io.github.resilience4j.retry.Retry;
import io.lettuce.core.pubsub.StatefulRedisPubSubConnection;
public class FaultTolerantPubSubConnection<K, V> extends AbstractFaultTolerantPubSubConnection<K, V, StatefulRedisPubSubConnection<K, V>> {
protected FaultTolerantPubSubConnection(final String name,
final StatefulRedisPubSubConnection<K, V> pubSubConnection,
final Retry retry) {
final StatefulRedisPubSubConnection<K, V> pubSubConnection) {
super(name, pubSubConnection, retry);
super(name, pubSubConnection);
}
}

4
service/src/main/java/org/whispersystems/textsecuregcm/redis/FaultTolerantRedisClient.java
View File

@ -147,13 +147,13 @@ public class FaultTolerantRedisClient {
final StatefulRedisPubSubConnection<String, String> pubSubConnection = redisClient.connectPubSub();
pubSubConnections.add(pubSubConnection);
return new FaultTolerantPubSubConnection<>(name, pubSubConnection, retry);
return new FaultTolerantPubSubConnection<>(name, pubSubConnection);
}
public FaultTolerantPubSubConnection<byte[], byte[]> createBinaryPubSubConnection() {
final StatefulRedisPubSubConnection<byte[], byte[]> pubSubConnection = redisClient.connectPubSub(ByteArrayCodec.INSTANCE);
pubSubConnections.add(pubSubConnection);
return new FaultTolerantPubSubConnection<>(name, pubSubConnection, retry);
return new FaultTolerantPubSubConnection<>(name, pubSubConnection);
}
}

2
service/src/main/java/org/whispersystems/textsecuregcm/redis/FaultTolerantRedisClusterClient.java
View File

@ -198,7 +198,7 @@ public class FaultTolerantRedisClusterClient {
final StatefulRedisClusterPubSubConnection<String, String> pubSubConnection = clusterClient.connectPubSub();
pubSubConnections.add(pubSubConnection);
return new FaultTolerantPubSubClusterConnection<>(name, pubSubConnection, retry, topologyChangedEventRetry,
return new FaultTolerantPubSubClusterConnection<>(name, pubSubConnection, topologyChangedEventRetry,
Schedulers.newSingle(name + "-redisPubSubEvents", true));
}

28
service/src/test/java/org/whispersystems/textsecuregcm/redis/FaultTolerantPubSubClusterConnectionTest.java
View File

@ -5,11 +5,8 @@
package org.whispersystems.textsecuregcm.redis;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.atLeastOnce;
import static org.mockito.Mockito.clearInvocations;
import static org.mockito.Mockito.mock;
@ -20,7 +17,6 @@ import static org.mockito.Mockito.when;
import io.github.resilience4j.core.IntervalFunction;
import io.github.resilience4j.retry.Retry;
import io.github.resilience4j.retry.RetryConfig;
import io.lettuce.core.RedisCommandTimeoutException;
import io.lettuce.core.RedisException;
import io.lettuce.core.cluster.event.ClusterTopologyChangedEvent;
import io.lettuce.core.cluster.pubsub.StatefulRedisClusterPubSubConnection;
@ -60,8 +56,6 @@ class FaultTolerantPubSubClusterConnectionTest {
retryConfiguration.setMaxAttempts(3);
retryConfiguration.setWaitDuration(10);
final Retry retry = Retry.of("test", retryConfiguration.toRetryConfig());
final RetryConfig resubscribeRetryConfiguration = RetryConfig.custom()
.maxAttempts(Integer.MAX_VALUE)
.intervalFunction(IntervalFunction.ofExponentialBackoff(5))
@ -69,27 +63,7 @@ class FaultTolerantPubSubClusterConnectionTest {
final Retry resubscribeRetry = Retry.of("test-resubscribe", resubscribeRetryConfiguration);
faultTolerantPubSubConnection = new FaultTolerantPubSubClusterConnection<>("test", pubSubConnection,
retry, resubscribeRetry, Schedulers.newSingle("test"));
}
@Test
void testRetry() {
when(pubSubCommands.get(anyString()))
.thenThrow(new RedisCommandTimeoutException())
.thenThrow(new RedisCommandTimeoutException())
.thenReturn("value");
assertEquals("value",
faultTolerantPubSubConnection.withPubSubConnection(connection -> connection.sync().get("key")));
when(pubSubCommands.get(anyString()))
.thenThrow(new RedisCommandTimeoutException())
.thenThrow(new RedisCommandTimeoutException())
.thenThrow(new RedisCommandTimeoutException())
.thenReturn("value");
assertThrows(RedisCommandTimeoutException.class,
() -> faultTolerantPubSubConnection.withPubSubConnection(connection -> connection.sync().get("key")));
resubscribeRetry, Schedulers.newSingle("test"));
}
@Nested