Upgrade to TC Core 5.10.8 & Platform 5.10.7 : Adopt new async client …

…implementation
ehcache · Feb 1, 2023 · 3e2082c · 3e2082c
1 parent 7818ce8
commit 3e2082c
Show file tree

Hide file tree

Showing 14 changed files with 199 additions and 146 deletions.
diff --git a/...main/java/org/ehcache/clustered/client/internal/SimpleClusterTierManagerClientEntity.java b/...main/java/org/ehcache/clustered/client/internal/SimpleClusterTierManagerClientEntity.java
@@ -30,11 +30,11 @@
 import org.terracotta.entity.EndpointDelegate;
 import org.terracotta.entity.EntityClientEndpoint;
 import org.terracotta.entity.EntityResponse;
-import org.terracotta.entity.InvokeFuture;
 import org.terracotta.entity.MessageCodecException;
-import org.terracotta.exception.EntityException;
 
 import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
 
 /**
  * The client-side {@link Entity} through which clustered cache operations are performed.
@@ -76,38 +76,38 @@ public void close() {
 
   @Override
   public void validate(ServerSideConfiguration config) throws ClusterException {
-    invokeInternal(messageFactory.validateStoreManager(config), false);
+    invokeInternal(messageFactory.validateStoreManager(config));
   }
 
   @Override
   public Set<String> prepareForDestroy() {
     try {
-      PrepareForDestroy response = (PrepareForDestroy) invokeInternal(messageFactory.prepareForDestroy(), true);
+      PrepareForDestroy response = (PrepareForDestroy) invokeInternal(messageFactory.prepareForDestroy());
       return response.getStores();
     } catch (ClusterException e) {
       // TODO handle this
     }
     return null;
   }
 
-  private EhcacheEntityResponse invokeInternal(EhcacheEntityMessage message, boolean replicate)
+  private EhcacheEntityResponse invokeInternal(EhcacheEntityMessage message)
       throws ClusterException {
 
     try {
-      EhcacheEntityResponse response = waitFor(invokeAsync(message, replicate));
+      EhcacheEntityResponse response = waitFor(invokeAsync(message));
       if (EhcacheResponseType.FAILURE.equals(response.getResponseType())) {
         throw ((Failure)response).getCause();
       } else {
         return response;
       }
-    } catch (EntityException | MessageCodecException e) {
-      throw new RuntimeException(message + " error: " + e.toString(), e);
+    } catch (ExecutionException | MessageCodecException e) {
+      throw new RuntimeException(message + " error: " + e, e);
     }
   }
 
-  private InvokeFuture<EhcacheEntityResponse> invokeAsync(EhcacheEntityMessage message, boolean replicate)
+  private Future<EhcacheEntityResponse> invokeAsync(EhcacheEntityMessage message)
       throws MessageCodecException {
-    return endpoint.beginInvoke().message(message).replicate(replicate).invoke();
+    return endpoint.message(message).invoke();
   }
 
   /**
@@ -117,10 +117,9 @@ private InvokeFuture<EhcacheEntityResponse> invokeAsync(EhcacheEntityMessage mes
    * @param future Future we want to get
    * @param <T> type of the response
    * @return the result of the get
-   * @throws EntityException exception that might be thrown by the future in case of error
    */
-  private static <T extends EntityResponse> T waitFor(InvokeFuture<T> future)
-      throws EntityException {
+  private static <T extends EntityResponse> T waitFor(Future<T> future)
+    throws ExecutionException {
     boolean interrupted = Thread.interrupted();
     try {
       while (true) {

diff --git a/.../src/main/java/org/ehcache/clustered/client/internal/lock/VoltronReadWriteLockClient.java b/.../src/main/java/org/ehcache/clustered/client/internal/lock/VoltronReadWriteLockClient.java
@@ -16,6 +16,8 @@
 
 package org.ehcache.clustered.client.internal.lock;
 
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
 import java.util.concurrent.Semaphore;
 import org.ehcache.clustered.common.internal.lock.LockMessaging;
 import org.ehcache.clustered.common.internal.lock.LockMessaging.LockOperation;
@@ -24,9 +26,7 @@
 import org.terracotta.connection.entity.Entity;
 import org.terracotta.entity.EndpointDelegate;
 import org.terracotta.entity.EntityClientEndpoint;
-import org.terracotta.entity.InvokeFuture;
 import org.terracotta.entity.MessageCodecException;
-import org.terracotta.exception.EntityException;
 
 public class VoltronReadWriteLockClient implements Entity {
 
@@ -106,26 +106,22 @@ private LockOperation getCurrentState() {
   }
 
   private LockTransition invoke(LockOperation operation) {
+    Future<LockTransition> result = endpoint.message(operation).invoke();
+    boolean interrupted = Thread.interrupted();
     try {
-      InvokeFuture<LockTransition> result = endpoint.beginInvoke().message(operation).replicate(false).invoke();
-      boolean interrupted = false;
-      try {
-        while (true) {
-          try {
-            return result.get();
-          } catch (InterruptedException ex) {
-            interrupted = true;
-          } catch (EntityException ex) {
-            throw new IllegalStateException(ex);
-          }
-        }
-      } finally {
-        if (interrupted) {
-          Thread.currentThread().interrupt();
+      while (true) {
+        try {
+          return result.get();
+        } catch (InterruptedException ex) {
+          interrupted = true;
+        } catch (ExecutionException ex) {
+          throw new IllegalStateException(ex.getCause());
         }
       }
-    } catch (MessageCodecException ex) {
-      throw new AssertionError(ex);
+    } finally {
+      if (interrupted) {
+        Thread.currentThread().interrupt();
+      }
     }
   }
 }
diff --git a/...e-client/src/main/java/org/ehcache/clustered/client/internal/service/ConnectionState.java b/...e-client/src/main/java/org/ehcache/clustered/client/internal/service/ConnectionState.java
@@ -41,7 +41,6 @@
 
 import java.io.IOException;
 import java.util.Properties;
-import java.util.Random;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.Executor;
@@ -50,6 +49,7 @@
 import java.util.concurrent.atomic.AtomicInteger;
 
 import static java.util.Objects.requireNonNull;
+import static org.ehcache.core.util.ExceptionUtil.containsCause;
 
 class ConnectionState {
 
@@ -114,9 +114,13 @@ public ClusterTierClientEntity createClusterTierClientEntity(String cacheId,
         break;
       } catch (EntityNotFoundException e) {
         throw new PerpetualCachePersistenceException("Cluster tier proxy '" + cacheId + "' for entity '" + entityIdentifier + "' does not exist.", e);
-      } catch (ConnectionClosedException | ConnectionShutdownException e) {
-        LOGGER.info("Disconnected from the server", e);
-        handleConnectionClosedException(true);
+      } catch (Throwable t) {
+        if (containsCause(t, ConnectionClosedException.class) || containsCause(t, ConnectionShutdownException.class)) {
+          LOGGER.info("Disconnected from the server", t);
+          handleConnectionClosedException(true);
+        } else {
+          throw t;
+        }
       }
     }
 
@@ -151,6 +155,12 @@ private void reconnect() {
         break;
       } catch (ConnectionClosedException | ConnectionException e) {
         LOGGER.error("Re-connection to server failed, trying again", e);
+      } catch (Throwable t) {
+        if (containsCause(t, ConnectionClosedException.class) || containsCause(t, ConnectionShutdownException.class)) {
+          LOGGER.error("Re-connection to server failed, trying again", t);
+        } else {
+          throw t;
+        }
       }
     }
   }
@@ -180,6 +190,14 @@ private boolean silentDestroyUtil() {
       LOGGER.info("Disconnected from the server", e);
       reconnect();
       return false;
+    } catch (Throwable t) {
+      if (containsCause(t, ConnectionClosedException.class) || containsCause(t, ConnectionShutdownException.class)) {
+        LOGGER.info("Disconnected from the server", t);
+        reconnect();
+        return false;
+      } else {
+        throw t;
+      }
     }
   }
 
@@ -262,6 +280,12 @@ public void destroyAll() throws CachePersistenceException {
         throw new CachePersistenceException("Cannot delete cluster tiers on " + connectionSource, e);
       } catch (ConnectionClosedException | ConnectionShutdownException e) {
         handleConnectionClosedException(false);
+      } catch (Throwable t) {
+        if (containsCause(t, ConnectionClosedException.class) || containsCause(t, ConnectionShutdownException.class)) {
+          handleConnectionClosedException(false);
+        } else {
+          throw t;
+        }
       }
     }
   }
@@ -285,6 +309,12 @@ public void destroy(String name) throws CachePersistenceException {
           }
         } catch (ConnectionClosedException | ConnectionShutdownException e) {
           reconnect();
+        } catch (Throwable t) {
+          if (containsCause(t, ConnectionClosedException.class) || containsCause(t, ConnectionShutdownException.class)) {
+            reconnect();
+          } else {
+            throw t;
+          }
         }
       }
 
@@ -299,6 +329,12 @@ public void destroy(String name) throws CachePersistenceException {
         break;
       } catch (ConnectionClosedException | ConnectionShutdownException e) {
         handleConnectionClosedException(false);
+      } catch (Throwable t) {
+        if (containsCause(t, ConnectionClosedException.class) || containsCause(t, ConnectionShutdownException.class)) {
+          handleConnectionClosedException(false);
+        } else {
+          throw t;
+        }
       }
     }
   }
@@ -315,6 +351,14 @@ private void autoCreateEntity() throws ClusterTierManagerValidationException, Il
         LOGGER.info("Disconnected from the server", e);
         reconnect();
         continue;
+      } catch (Throwable t) {
+        if (containsCause(t, ConnectionClosedException.class) || containsCause(t, ConnectionShutdownException.class)) {
+          LOGGER.info("Disconnected from the server", t);
+          reconnect();
+          continue;
+        } else {
+          throw t;
+        }
       }
 
       try {
@@ -330,6 +374,13 @@ private void autoCreateEntity() throws ClusterTierManagerValidationException, Il
       } catch (ConnectionClosedException | ConnectionShutdownException e) {
         LOGGER.info("Disconnected from the server", e);
         reconnect();
+      } catch (Throwable t) {
+        if (containsCause(t, ConnectionClosedException.class) || containsCause(t, ConnectionShutdownException.class)) {
+          LOGGER.info("Disconnected from the server", t);
+          reconnect();
+        } else {
+          throw t;
+        }
       }
     }
 
@@ -349,8 +400,12 @@ private void handleConnectionClosedException(boolean retrieve) throws ClusterTie
         }
         connectionRecoveryListener.run();
         break;
-      } catch (ConnectionClosedException | ConnectionShutdownException e) {
-        LOGGER.info("Disconnected from the server", e);
+      } catch (Throwable t) {
+        if (containsCause(t, ConnectionClosedException.class) || containsCause(t, ConnectionShutdownException.class)) {
+          LOGGER.info("Disconnected from the server", t);
+        } else {
+          throw t;
+        }
       }
     }
   }

diff --git a/...ent/src/main/java/org/ehcache/clustered/client/internal/store/CommonServerStoreProxy.java b/...ent/src/main/java/org/ehcache/clustered/client/internal/store/CommonServerStoreProxy.java
@@ -143,7 +143,7 @@ public ChainEntry get(long key) throws TimeoutException {
   @Override
   public void append(long key, ByteBuffer payLoad) {
     try {
-      entity.invokeAndWaitForReceive(new AppendMessage(key, payLoad), true);
+      entity.invokeAndWaitForComplete(new AppendMessage(key, payLoad), true);
     } catch (Exception e) {
       throw new ServerStoreProxyException(e);
     }

diff --git a/...c/main/java/org/ehcache/clustered/client/internal/store/ReconnectingServerStoreProxy.java b/...c/main/java/org/ehcache/clustered/client/internal/store/ReconnectingServerStoreProxy.java
@@ -16,7 +16,6 @@
 package org.ehcache.clustered.client.internal.store;
 
 import org.ehcache.clustered.client.internal.store.lock.LockingServerStoreProxy;
-import org.ehcache.clustered.client.internal.store.lock.LockingServerStoreProxyImpl;
 import org.ehcache.clustered.common.internal.store.Chain;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -29,6 +28,8 @@
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicReference;
 
+import static org.ehcache.core.util.ExceptionUtil.containsCause;
+
 public class ReconnectingServerStoreProxy implements LockingServerStoreProxy {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(ReconnectingServerStoreProxy.class);
@@ -54,8 +55,12 @@ public String getCacheId() {
   public void close() {
     try {
       proxy().close();
-    } catch (ConnectionClosedException | ConnectionShutdownException e) {
-      LOGGER.debug("Store was already closed, since connection was closed");
+    } catch (Throwable t) {
+      if (containsCause(t, ConnectionClosedException.class) || containsCause(t, ConnectionShutdownException.class)) {
+        LOGGER.debug("Store was already closed, since connection was closed");
+      } else {
+        throw t;
+      }
     }
   }
 
@@ -119,7 +124,7 @@ private <T> T onStoreProxy(TimeoutExceptionFunction<LockingServerStoreProxy, T>
     try {
       return function.apply(storeProxy);
     } catch (ServerStoreProxyException sspe) {
-      if (sspe.getCause() instanceof ConnectionClosedException) {
+      if (containsCause(sspe, ConnectionClosedException.class)) {
         if (delegateRef.compareAndSet(storeProxy, new ReconnectInProgressProxy(storeProxy.getCacheId()))) {
           onReconnect.run();
         }