chore: improve subscription reconnection logic (#435)

Move the topic subscription reconnection logic into a retry strategy
and eligibility strategy that can be configured by the client.

Add the subscription retry strategy to TopicConfiguration. Add new
constructors that take the retry strategy, and new ones that don't take
a logger, which is only used inside the TopicConfiguration itself.

Prevent potential race conditions that could occur when unsubscribing
from a connection while it is reconnecting by checking if the user
called unsubscribe before reconnecting, and by making the variables
used in reconnection atomic.

Call close in SubscriptionWrapper when a stream ends to stop a
subscription from leaking threads from the retry executor service. We
may want to share a thread pool between the subscriptions instead of
making a new one per subscription, so that we don't need to make a new
one each time. We could also use it to execute the callbacks, which are
currently executed in the gRPC thread pool and could cause backpressure
if they are long-running.

Add empty default implementations for the test methods in
IScsTopicConnection to clean up its use in the topic client.

Remove SendSubscribeOptions, since we already have a class that contains
the callbacks, and because it had an unused subscription object.

Get rid of CancelableClientCallStreamObserver's ClientCallStreamObserver
extension, because that class is meant for observing requests, not
responses, and we only need a simple cancel method.

Make TopicClientLocalTest's unrecoverable error 'not found', because the
previous error is now recoverable.

Unsubscribe should end with onCompleted instead of onError

---------

Co-authored-by: anitarua <anita@momentohq.com>

Author: nand4011

momento-sdk/src/intTest/java/momento/sdk/retry/TopicClientLocalTest.java

@@ -119,7 +119,7 @@ void testSubscribe_shouldNotRetryWithUnrecoverableError() throws Exception {
     final int streamErrorMessageLimit = 3;
     final MomentoLocalMiddlewareArgs momentoLocalMiddlewareArgs =
         new MomentoLocalMiddlewareArgs.Builder(logger, UUID.randomUUID().toString())
-            .streamError(MomentoErrorCode.INTERNAL_SERVER_ERROR)
+            .streamError(MomentoErrorCode.NOT_FOUND_ERROR)
             .streamErrorRpcList(Collections.singletonList(MomentoRpcMethod.TOPIC_SUBSCRIBE))
             .streamErrorMessageLimit(streamErrorMessageLimit)
             .build();

momento-sdk/src/main/java/momento/sdk/CancelableClientCallStreamObserver.java

@@ -5,36 +5,18 @@
 import javax.annotation.Nullable;
 
 public abstract class CancelableClientCallStreamObserver<TResp>
-    extends ClientCallStreamObserver<TResp> implements ClientResponseObserver<Object, TResp> {
+    implements ClientResponseObserver<Object, TResp> {
 
-  private ClientCallStreamObserver requestStream;
+  private ClientCallStreamObserver<Object> requestStream;
 
-  @Override
-  public boolean isReady() {
-    return false;
-  }
-
-  @Override
-  public void setOnReadyHandler(Runnable onReadyHandler) {}
-
-  @Override
-  public void request(int count) {}
-
-  @Override
-  public void setMessageCompression(boolean enable) {}
-
-  @Override
-  public void disableAutoInboundFlowControl() {}
-
-  @Override
   public void cancel(@Nullable String message, @Nullable Throwable cause) {
     if (requestStream != null) {
       requestStream.cancel(message, cause);
     }
   }
 
   @Override
-  public void beforeStart(ClientCallStreamObserver requestStream) {
+  public void beforeStart(ClientCallStreamObserver<Object> requestStream) {
     this.requestStream = requestStream;
   }
 }

momento-sdk/src/main/java/momento/sdk/IScsTopicConnection.java

@@ -12,15 +12,15 @@ interface IScsTopicConnection {
    * <p>Note: This method is intended for testing purposes and should never be called from outside
    * of tests.
    */
-  void close();
+  default void close() {}
 
   /**
    * Opens the connection.
    *
    * <p>Note: This method is intended for testing purposes and should never be called from outside
    * of tests.
    */
-  void open();
+  default void open() {}
 
   /**
    * Subscribes to a specific topic using the provided subscription request and observer.

momento-sdk/src/main/java/momento/sdk/ScsTopicClient.java

@@ -2,9 +2,8 @@
 
 import com.google.protobuf.ByteString;
 import grpc.cache_client.pubsub._PublishRequest;
-import grpc.cache_client.pubsub._SubscriptionItem;
-import grpc.cache_client.pubsub._SubscriptionRequest;
 import grpc.cache_client.pubsub._TopicValue;
+import grpc.common._Empty;
 import io.grpc.stub.StreamObserver;
 import java.util.concurrent.CompletableFuture;
 import javax.annotation.Nonnull;
@@ -14,19 +13,19 @@
 import momento.sdk.internal.SubscriptionState;
 import momento.sdk.responses.topic.TopicPublishResponse;
 import momento.sdk.responses.topic.TopicSubscribeResponse;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import momento.sdk.retry.SubscriptionRetryStrategy;
 
 public class ScsTopicClient extends ScsClientBase {
 
-  private final Logger logger = LoggerFactory.getLogger(ScsTopicClient.class);
   private final ScsTopicGrpcStubsManager topicGrpcStubsManager;
   private final long DEFAULT_REQUEST_TIMEOUT_SECONDS = 5;
+  private final SubscriptionRetryStrategy subscriptionRetryStrategy;
 
   public ScsTopicClient(
       @Nonnull CredentialProvider credentialProvider, @Nonnull TopicConfiguration configuration) {
     super(null);
     this.topicGrpcStubsManager = new ScsTopicGrpcStubsManager(credentialProvider, configuration);
+    this.subscriptionRetryStrategy = configuration.getSubscriptionRetryStrategy();
   }
 
   public CompletableFuture<TopicPublishResponse> publish(
@@ -60,7 +59,7 @@ public CompletableFuture<TopicPublishResponse> publish(
   }
 
   public CompletableFuture<TopicSubscribeResponse> subscribe(
-      String cacheName, String topicName, ISubscriptionCallbacks options) {
+      String cacheName, String topicName, ISubscriptionCallbacks callbacks) {
     try {
       ValidationUtils.checkCacheNameValid(cacheName);
       ValidationUtils.checkTopicNameValid(topicName);
@@ -69,24 +68,7 @@ public CompletableFuture<TopicSubscribeResponse> subscribe(
           new TopicSubscribeResponse.Error(CacheServiceExceptionMapper.convert(e)));
     }
 
-    SubscriptionState subscriptionState = new SubscriptionState();
-    TopicSubscribeResponse.Subscription subscription =
-        new TopicSubscribeResponse.Subscription(subscriptionState);
-    SendSubscribeOptions sendSubscribeOptions =
-        new SendSubscribeOptions(
-            cacheName,
-            topicName,
-            options::onItem,
-            options::onCompleted,
-            options::onError,
-            options::onDiscontinuity,
-            options::onHeartbeat,
-            options::onConnectionLost,
-            options::onConnectionRestored,
-            subscriptionState,
-            subscription);
-
-    return sendSubscribe(sendSubscribeOptions);
+    return sendSubscribe(cacheName, topicName, callbacks);
   }
 
   private CompletableFuture<TopicPublishResponse> sendPublish(
@@ -105,10 +87,10 @@ private CompletableFuture<TopicPublishResponse> sendPublish(
           .getNextUnaryStub()
           .publish(
               request,
-              new StreamObserver() {
+              new StreamObserver<_Empty>() {
 
                 @Override
-                public void onNext(Object value) {
+                public void onNext(_Empty value) {
                   // Do nothing
                 }
 
@@ -133,28 +115,12 @@ public void onCompleted() {
   }
 
   private CompletableFuture<TopicSubscribeResponse> sendSubscribe(
-      SendSubscribeOptions sendSubscribeOptions) {
-    SubscriptionWrapper subscriptionWrapper;
-
-    IScsTopicConnection connection =
-        new IScsTopicConnection() {
-          @Override
-          public void close() {
-            logger.warn("Closing the connection (for testing purposes only)");
-          }
+      String cacheName, String topicName, ISubscriptionCallbacks callbacks) {
+    final SubscriptionState subscriptionState = new SubscriptionState();
 
-          @Override
-          public void open() {
-            logger.warn("Opening the connection (for testing purposes only)");
-          }
-
-          @Override
-          public void subscribe(
-              _SubscriptionRequest subscriptionRequest,
-              CancelableClientCallStreamObserver<_SubscriptionItem> subscription) {
-            topicGrpcStubsManager.getNextStreamStub().subscribe(subscriptionRequest, subscription);
-          }
-        };
+    final IScsTopicConnection connection =
+        (request, subscription) ->
+            topicGrpcStubsManager.getNextStreamStub().subscribe(request, subscription);
 
     long configuredTimeoutSeconds =
         topicGrpcStubsManager
@@ -166,18 +132,24 @@ public void subscribe(
     long firstMessageSubscribeTimeoutSeconds =
         configuredTimeoutSeconds > 0 ? configuredTimeoutSeconds : DEFAULT_REQUEST_TIMEOUT_SECONDS;
 
-    subscriptionWrapper =
+    @SuppressWarnings("resource") // the wrapper closes itself when a subscription ends.
+    final SubscriptionWrapper subscriptionWrapper =
         new SubscriptionWrapper(
-            connection, sendSubscribeOptions, firstMessageSubscribeTimeoutSeconds);
+            cacheName,
+            topicName,
+            connection,
+            callbacks,
+            subscriptionState,
+            firstMessageSubscribeTimeoutSeconds,
+            subscriptionRetryStrategy);
     final CompletableFuture<Void> subscribeFuture = subscriptionWrapper.subscribeWithRetry();
     return subscribeFuture.handle(
         (v, ex) -> {
           if (ex != null) {
             return new TopicSubscribeResponse.Error(CacheServiceExceptionMapper.convert(ex));
           } else {
-            sendSubscribeOptions.subscriptionState.setUnsubscribeFn(
-                subscriptionWrapper::unsubscribe);
-            return new TopicSubscribeResponse.Subscription(sendSubscribeOptions.subscriptionState);
+            subscriptionState.setUnsubscribeFn(subscriptionWrapper::unsubscribe);
+            return new TopicSubscribeResponse.Subscription(subscriptionState);
           }
         });
   }