FISH-105 Migrate EJB Timers from Live Instances

appserver/ejb/ejb-container/src/main/java/org/glassfish/ejb/admin/cli/MigrateTimers.java

@@ -37,7 +37,7 @@
  * only if the new code is made subject to such option by the copyright
  * holder.
  */
-// Portions Copyright [2018] [Payara Foundation and/or its affiliates]
+// Portions Copyright [2018-2021] [Payara Foundation and/or its affiliates]
 
 package org.glassfish.ejb.admin.cli;
 
@@ -196,18 +196,12 @@ private String validateCluster() {
             return localStrings.getString("migrate.timers.fromServerNotClusteredInstance", fromServer);
         }
 
-        //verify fromServer is not running
-        if (isServerRunning(fromServer)) {
-            return localStrings.getString(
-                    "migrate.timers.migrateFromServerStillRunning", fromServer);
-        }
-
         //if destinationServer is not set, or set to DAS, pick a running instance
         //in the same cluster as fromServer
         if(target.equals(SystemPropertyConstants.DEFAULT_SERVER_INSTANCE_NAME)) {
             List<Server> instances = fromServerCluster.getInstances();
             for(Server instance : instances) {
-                if(instance.isRunning()) {
+                if(instance.isRunning() && !instance.getName().equals(fromServer)) {
                     target = instance.getName();
                     needRedirect = true;
                 }
@@ -241,18 +235,12 @@ private String validateDG() {
             return localStrings.getString("migrate.timers.fromServerNotDG", fromServer);
         }
 
-        //verify fromServer is not running
-        if (isServerRunning(fromServer)) {
-            return localStrings.getString(
-                    "migrate.timers.migrateFromServerStillRunning", fromServer);
-        }
-
         //if destinationServer is not set, or set to DAS, pick a running instance
         //in the same cluster as fromServer
         if(target.equals(SystemPropertyConstants.DEFAULT_SERVER_INSTANCE_NAME)) {
             List<Server> instances = dgs.get(0).getInstances();
             for(Server instance : instances) {
-                if(instance.isRunning()) {
+                if(instance.isRunning() && !instance.getName().equals(fromServer)) {
                     target = instance.getName();
                     needRedirect = true;
                 }

appserver/ejb/ejb-container/src/main/resources/org/glassfish/ejb/admin/cli/LocalStrings.properties

@@ -37,10 +37,9 @@
 # only if the new code is made subject to such option by the copyright
 # holder.
 #
-# Portions Copyright [2018] [Payara Foundation and/or its affiliates]
+# Portions Copyright [2018-2021] [Payara Foundation and/or its affiliates]
 migrate.timers.count=Migrated {0} timers from {1} to {2}.
 migrate.timers.fromServerNotClusteredInstance=The server from which to migrate timers is not part of any cluster: {0}.
-migrate.timers.migrateFromServerStillRunning=The server from which to migrate timers is still running: {0}.
 migrate.timers.fromServerAndTargetNotInSameCluster=The server from which to migrate timers: {0}, is not in the same cluster as the target server: {1}.
 migrate.timers.destinationServerIsNotAlive=The target server is not running: {0}.
 migrate.timers.noRunningInstanceToChoose=No running instance is available to replace the specified target: {0}.

appserver/ejb/ejb-full-container/src/main/java/org/glassfish/ejb/persistent/timer/PersistentEJBTimerService.java

@@ -37,7 +37,7 @@
  * only if the new code is made subject to such option by the copyright
  * holder.
  */
-// Portions Copyright [2016-2018] [Payara Foundation and/or its affiliates]
+// Portions Copyright [2016-2021] [Payara Foundation and/or its affiliates]
 
 package org.glassfish.ejb.persistent.timer;
 
@@ -221,7 +221,7 @@ public int migrateTimers(String fromOwnerId) {
         TransactionManager tm = ejbContainerUtil.getTransactionManager();
 
         Set toRestore = null;
-	int totalTimersMigrated = 0;
+	    int totalTimersMigrated = 0;
 
         try {
 
@@ -1234,13 +1234,11 @@ private TimerState getValidTimerFromDB(TimerPrimaryKey timerId) {
                 // the current server
                 if( ! ( timer.getOwnerId().equals(
                         ownerIdOfThisServer_) ) ) {
-                    logger.log(Level.WARNING,
+                    logger.log(Level.INFO,
                             "The timer (" + timerId + ") is not owned by " +
                             "server (" + ownerIdOfThisServer_ + ") that " +
                             "initiated the ejbTimeout. This timer is now " +
-                            "owned by (" + timer.getOwnerId() + "). \n" +
-                            "Hence delete the timer from " +
-                            ownerIdOfThisServer_ + "'s cache.");
+                            "owned by (" + timer.getOwnerId() + ").");
 
                     result = false;
                 }

appserver/payara-appserver-modules/hazelcast-ejb-timer/pom.xml

@@ -3,7 +3,7 @@
 
   DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.
 
-  Copyright (c) 2014-2019 Payara Foundation and/or its affiliates. All rights reserved.
+  Copyright (c) 2014-2021 Payara Foundation and/or its affiliates. All rights reserved.
 
   The contents of this file are subject to the terms of either the GNU
   General Public License Version 2 only ("GPL") or the Common Development
@@ -82,5 +82,10 @@
             <groupId>org.mockito</groupId>
             <artifactId>mockito-core</artifactId>
         </dependency>
+        <dependency>
+            <groupId>fish.payara.server.internal.payara-appserver-modules</groupId>
+            <artifactId>payara-micro-service</artifactId>
+            <version>${project.version}</version>
+        </dependency>
     </dependencies>
 </project>

appserver/payara-appserver-modules/hazelcast-ejb-timer/src/main/java/fish/payara/ejb/timer/hazelcast/EjbTimerEvent.java

@@ -0,0 +1,78 @@
+/*
+ * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.
+ *
+ * Copyright (c) 2021 Payara Foundation and/or its affiliates. All rights reserved.
+ *
+ * The contents of this file are subject to the terms of either the GNU
+ * General Public License Version 2 only ("GPL") or the Common Development
+ * and Distribution License("CDDL") (collectively, the "License").  You
+ * may not use this file except in compliance with the License.  You can
+ * obtain a copy of the License at
+ * https://github.com/payara/Payara/blob/master/LICENSE.txt
+ * See the License for the specific
+ * language governing permissions and limitations under the License.
+ *
+ * When distributing the software, include this License Header Notice in each
+ * file and include the License file at glassfish/legal/LICENSE.txt.
+ *
+ * GPL Classpath Exception:
+ * The Payara Foundation designates this particular file as subject to the "Classpath"
+ * exception as provided by the Payara Foundation in the GPL Version 2 section of the License
+ * file that accompanied this code.
+ *
+ * Modifications:
+ * If applicable, add the following below the License Header, with the fields
+ * enclosed by brackets [] replaced by your own identifying information:
+ * "Portions Copyright [year] [name of copyright owner]"
+ *
+ * Contributor(s):
+ * If you wish your version of this file to be governed by only the CDDL or
+ * only the GPL Version 2, indicate your decision by adding "[Contributor]
+ * elects to include this software in this distribution under the [CDDL or GPL
+ * Version 2] license."  If you don't indicate a single choice of license, a
+ * recipient has the option to distribute your version of this file under
+ * either the CDDL, the GPL Version 2 or to extend the choice of license to
+ * its licensees as provided above.  However, if you add GPL Version 2 code
+ * and therefore, elected the GPL Version 2 license, then the option applies
+ * only if the new code is made subject to such option by the copyright
+ * holder.
+ */
+package fish.payara.ejb.timer.hazelcast;
+
+import fish.payara.micro.data.InstanceDescriptor;
+import java.io.Serializable;
+
+/**
+ * Class and enum for sending EJB timer events across the Hazelcast Datagrid.
+ */
+public class EjbTimerEvent implements Serializable {
+    private static final long serialVersionUID = 1L;
+
+    public static final String EJB_TIMER_EVENTS_TOPIC = "payara.server.internal.ejb.timer.event";
+
+    private final Event eventType;
+    private final InstanceDescriptor id;
+
+    /**
+     *
+     * @param eventType The type of EJB Timer event
+     * @param id The InstanceDescriptor pertaining to the EJB Timer event
+     */
+    public EjbTimerEvent(Event eventType, InstanceDescriptor id) {
+        this.eventType = eventType;
+        this.id = id;
+    }
+
+    public Event getEventType() {
+        return eventType;
+    }
+
+    public InstanceDescriptor getId() {
+        return id;
+    }
+
+    public enum Event {
+
+        MIGRATED
+    }
+}

appserver/payara-appserver-modules/hazelcast-ejb-timer/src/main/java/fish/payara/ejb/timer/hazelcast/HazelcastTimerStore.java

@@ -1,7 +1,7 @@
 /*
  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.
  *
- * Copyright (c) 2016-2020 Payara Foundation and/or its affiliates. All rights reserved.
+ * Copyright (c) 2016-2021 Payara Foundation and/or its affiliates. All rights reserved.
  *
  * The contents of this file are subject to the terms of either the GNU
  * General Public License Version 2 only ("GPL") or the Common Development
@@ -50,9 +50,13 @@
 import com.sun.ejb.containers.TimerPrimaryKey;
 import com.sun.enterprise.deployment.MethodDescriptor;
 import com.sun.logging.LogDomains;
+import fish.payara.appserver.micro.services.PayaraInstanceImpl;
+import fish.payara.micro.data.InstanceDescriptor;
 import fish.payara.nucleus.cluster.ClusterListener;
 import fish.payara.nucleus.cluster.MemberEvent;
 import fish.payara.nucleus.cluster.PayaraCluster;
+import fish.payara.nucleus.eventbus.ClusterMessage;
+import fish.payara.nucleus.eventbus.MessageReceiver;
 import fish.payara.nucleus.hazelcast.HazelcastCore;
 import java.io.Serializable;
 import java.lang.reflect.Method;
@@ -81,7 +85,8 @@
  * @author steve
  * @since 4.1.1.163
  */
-public class HazelcastTimerStore extends NonPersistentEJBTimerService implements ClusterListener {
+public class HazelcastTimerStore extends NonPersistentEJBTimerService implements ClusterListener,
+        MessageReceiver<EjbTimerEvent> {
 
     private static final String EJB_TIMER_CACHE_NAME = "HZEjbTmerCache";
     private static final String EJB_TIMER_CONTAINER_CACHE_NAME = "HZEjbTmerContainerCache";
@@ -100,7 +105,8 @@ static void init(HazelcastCore core) {
             HazelcastTimerStore store = new HazelcastTimerStore(core);
             Globals.getDefaultBaseServiceLocator().getService(PayaraCluster.class).addClusterListener(store);
             EJBTimerService.setPersistentTimerService(store);
-
+            Globals.getDefaultBaseServiceLocator().getService(PayaraCluster.class).getEventBus().addMessageReceiver(
+                    EjbTimerEvent.EJB_TIMER_EVENTS_TOPIC, store);
         } catch (Exception ex) {
             Logger.getLogger(HazelcastTimerStore.class.getName()).log(Level.WARNING, "Problem when initialising Timer Store", ex);
         }
@@ -475,6 +481,8 @@ protected boolean isValidTimerForThisServer(TimerPrimaryKey timerId, RuntimeTime
             HZTimer timer = pkCache.get(timerId.timerId);
             if (timer == null || !timer.getMemberName().equals(serverName)) {
                 result = false;
+
+                removeLocalTimer(timer);
             }
         }
         return result;
@@ -516,13 +524,9 @@ public int migrateTimers(String fromOwnerId) {
         String ownerIdOfThisServer = getOwnerIdOfThisServer();
 
         if (fromOwnerId.equals(ownerIdOfThisServer)) {
-            /// Error. The server from which timers are being
-            // migrated should never be up and running OR receive this
-            // notification.
-            logger.log(Level.WARNING, "Attempt to migrate timers from an active server instance {0}", ownerIdOfThisServer);
-            throw new IllegalStateException("Attempt to migrate timers from "
-                    + " an active server instance "
-                    + ownerIdOfThisServer);
+            logger.log(Level.WARNING, "Attempt to migrate timers from {0} to itself",
+                    ownerIdOfThisServer);
+            throw new IllegalStateException("Attempt to migrate timers from " + ownerIdOfThisServer + " to itself");
         }
 
         logger.log(Level.INFO, "Beginning timer migration process from owner {0} to {1}", new Object[]{fromOwnerId, ownerIdOfThisServer});
@@ -546,7 +550,6 @@ public int migrateTimers(String fromOwnerId) {
             totalTimersMigrated++;
         }
 
-// XXX if( totalTimersMigrated  == toRestore.size() ) { XXX ???
         if (totalTimersMigrated > 0) {
 
             boolean success = false;
@@ -560,6 +563,8 @@ public int migrateTimers(String fromOwnerId) {
                 _restoreTimers(toRestore.values());
                 success = true;
 
+                // Inform fromServer that timers have been migrated and it needs to clear its local cache
+                _notifyMigratedFromInstance(fromOwnerId);
             } catch (Exception e) {
 
                 logger.log(Level.FINE, "timer restoration error", e);
@@ -685,6 +690,10 @@ protected void resetLastExpiration(TimerPrimaryKey timerId, RuntimeTimerState ti
                 return;
             }
 
+            if (removeLocalTimer(timer)) {
+                return;
+            }
+
             Date now = new Date();
             timer.setLastExpiration(now);
             pkCache.put(timer.getKey().timerId, timer);
@@ -694,7 +703,7 @@ protected void resetLastExpiration(TimerPrimaryKey timerId, RuntimeTimerState ti
             // enabled.
             // @@@ add configuration for update-db-on-delivery
             if (logger.isLoggable(Level.FINE)) {
-                logger.log(Level.FINE, "Setting last expiration  for periodic timer {0} to {1}", new Object[]{timerState, now});
+                logger.log(Level.FINE, "Setting last expiration for periodic timer {0} to {1}", new Object[]{timerState, now});
             }
         }
     }
@@ -736,8 +745,7 @@ protected boolean timerExists(TimerPrimaryKey timerId) {
     }
 
     @Override
-    protected void stopTimers(long containerId
-    ) {
+    protected void stopTimers(long containerId) {
         super.stopTimers(containerId);
         stopTimers(containerCache.get(containerId));
     }
@@ -1194,4 +1202,81 @@ public void memberRemoved(MemberEvent event) {
         }
     }
 
+    /**
+     * Remove all local timers that are no longer owned by this instance.
+     */
+    private void removeLocalTimers() {
+        Collection<HZTimer> allTimers = pkCache.values();
+        for (HZTimer timer : allTimers) {
+            removeLocalTimer(timer);
+        }
+    }
+
+    /**
+     * Removes a given local timer if it is no longer owned by this instance
+     * @param timer The timer to potentially be removed.
+     * @return True if a timer was removed.
+     */
+    private boolean removeLocalTimer(HZTimer timer) {
+        boolean result = false;
+        TimerPrimaryKey timerId = timer.getKey();
+
+        if (!timer.getOwnerId().equals(getOwnerIdOfThisServer()) && getTimerState(timerId) != null) {
+            logger.log(Level.INFO,
+                    "The timer (" + timerId + ") is now owned by (" + timer.getOwnerId() + "). Removing from " +
+                            "local cache");
+
+            // We don't want to expunge it from the Hazelcast caches since it's a distributed cache,
+            // so only expunge from local cache
+            super.expungeTimer(timerId, false);
+            result = true;
+        }
+
+        return result;
+    }
+
+    /**
+     * Sends an {@link EjbTimerEvent} across the DataGrid with the {@link InstanceDescriptor} of the instance from
+     * which the EJB timers were migrated from, to allow other instances to react to the migration.
+     * @param fromOwnerId The {@link InstanceDescriptor} of the instance from which the timers were migrated from
+     */
+    private void _notifyMigratedFromInstance(String fromOwnerId) {
+        PayaraCluster cluster = Globals.getDefaultBaseServiceLocator().getService(PayaraCluster.class);
+        PayaraInstanceImpl instance = Globals.getDefaultBaseServiceLocator().getService(PayaraInstanceImpl.class);
+
+        if (cluster == null || instance == null) {
+            return;
+        }
+
+        // Get the InstanceDescriptor of the fromOwnerId instance
+        InstanceDescriptor fromOwnerInstanceDescriptor = null;
+        for (InstanceDescriptor instanceDescriptor : instance.getClusteredPayaras()) {
+            if (instanceDescriptor.getInstanceName().equals(fromOwnerId)) {
+                fromOwnerInstanceDescriptor = instanceDescriptor;
+                break;
+            }
+        }
+
+        if (fromOwnerInstanceDescriptor == null) {
+            return;
+        }
+
+        EjbTimerEvent ejbTimerEvent = new EjbTimerEvent(EjbTimerEvent.Event.MIGRATED, fromOwnerInstanceDescriptor);
+        ClusterMessage<EjbTimerEvent> message = new ClusterMessage<>(ejbTimerEvent);
+        cluster.getEventBus().publish(EjbTimerEvent.EJB_TIMER_EVENTS_TOPIC, message);
+    }
+
+    @Override
+    public void receiveMessage(ClusterMessage<EjbTimerEvent> ejbTimerEvent) {
+        if (ejbTimerEvent.getPayload().getEventType().equals(EjbTimerEvent.Event.MIGRATED)) {
+            PayaraInstanceImpl instance = Globals.getDefaultBaseServiceLocator().getService(PayaraInstanceImpl.class);
+            if (instance == null) {
+                return;
+            }
+
+            if (ejbTimerEvent.getPayload().getId().equals(instance.getLocalDescriptor())) {
+                removeLocalTimers();
+            }
+        }
+    }
 }