Fixed issues with event processing in cluster nodes.

- Repeated failed to process events were logged because the head of
the event queue wasn't being removed. Instead, the queue was traversed
through its iterator and the Iterator.remove() were being called only
upon a success. Hence further processing tasks got the same event that
kept failing.

- It was possible that a REMOVE event was queued while its corresponding
ADD event wasn't yet processed, leading to unnecessary logging of a failed
ADD event, since the object was already deleted from the catalog. Added a
check to verify whether a REMOVE event for the same object is scheduled
before complaining when processing a ADD/CHANGE event whose object is not
found in the catalog.

- The Hazelcast synchronizer was not aware of application lifecycle, so
when a node joined the cluster it starts trying to process events before
the application is fully loaded, and keeps processing events when the
application is shuttting down. Added a 'started' state to HzSynchronizer
that's set to true once the app is ready and to false when it's shutting down,
preventing the attempt to process events when the catalog is in a state of
flux, which is specially problematic with jdbcconfig. This avoids
ConcurrentModificationExceptions to be thrown while the synchronizer tries
to notify the catalog listeners which are being added/removed, and hence
may try to notify a catalog listener that's no longer valid.

- Homogeinized the log messages to contain the current and origin node ids
and a consistent representation of the events.

- Guarded the traversal of catalog listeners using a local copy since the
GeoSever's CatalogImpl.listeners list is not thread safe.

src/main/java/org/geoserver/cluster/ConfigChangeEvent.java

@@ -1,5 +1,7 @@
 package org.geoserver.cluster;
 
+import static com.google.common.base.Objects.equal;
+import java.io.Serializable;
 import java.util.HashMap;
 import java.util.Map;
 
@@ -42,6 +44,8 @@
 import org.geoserver.config.impl.LoggingInfoImpl;
 import org.geoserver.config.impl.SettingsInfoImpl;
 
+import com.google.common.base.Objects;
+
 /**
  * Event for 
  * @author Justin Deoliveira, OpenGeo
@@ -113,6 +117,45 @@ public ConfigChangeEvent(String id, String name, Class<? extends Info> clazz, Ty
         this.type = type;
     }
 
+    @Override
+    public String toString() {
+        StringBuilder sb = new StringBuilder(String.valueOf(type)).append(" ");
+
+        Serializable source = getSource();
+        if (source != null) {
+            sb.append('(').append(source).append(") ");
+        }
+
+        sb.append("[id:").append(id).append(", name:").append(name).append("]");
+        return sb.toString();
+    }
+
+    /**
+     * Equals is based on {@link #getObjectId() id}, {@link #getObjectName() name}, and
+     * {@link #getChangeType() changeType}. {@link #getObjectClass() class} is left off because it
+     * can be a proxy class and id/name/type are good enough anyways (given ids are unique, no two
+     * objects of different class can have the same id).
+     */
+    @Override
+    public boolean equals(Object o) {
+        if (!(o instanceof ConfigChangeEvent)) {
+            return false;
+        }
+        ConfigChangeEvent e = (ConfigChangeEvent) o;
+        return equal(id, e.id) && equal(type, e.type);
+    }
+
+    /**
+     * Hash code is based on {@link #getObjectId() id}, {@link #getObjectName() name}, and
+     * {@link #getChangeType() changeType}. {@link #getObjectClass() class} is left off because it
+     * can be a proxy class and id/name/type are good enough anyways (given ids are unique, no two
+     * objects of different class can have the same id).
+     */
+    @Override
+    public int hashCode() {
+        return Objects.hashCode(ConfigChangeEvent.class, id, name, type);
+    }
+
     public String getObjectId() {
         return id;
     }

src/main/java/org/geoserver/cluster/Event.java

@@ -1,6 +1,9 @@
 package org.geoserver.cluster;
 
 import java.io.Serializable;
+import java.util.concurrent.atomic.AtomicLong;
+
+import com.google.common.base.Objects;
 
 /**
  * Base class for events to be signalled across the cluster.  Carries an identifier for the 
@@ -10,9 +13,9 @@ public class Event implements Serializable {
 
     /** serialVersionUID */
     private static final long serialVersionUID = 1L;
-
+    
     Serializable source;
-
+    
     /**
      * Set an identifier for the node on which the event originates.
      * @param source
@@ -29,4 +32,21 @@ public Serializable getSource() {
         return source;
     }
 
+    @Override
+    public String toString() {
+        return new StringBuilder("Event[").append(source).append(']').toString();
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hashCode(Event.class, source);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (!(o instanceof Event)) {
+            return false;
+        }
+        return Objects.equal(source, ((Event) o).source);
+    }
 }

src/main/java/org/geoserver/cluster/GeoServerSynchronizer.java

@@ -17,7 +17,7 @@ public class GeoServerSynchronizer extends ConfigurationListenerAdapter
 
     protected ClusterConfigWatcher configWatcher;
 
-    public final void initialize(ClusterConfigWatcher configWatcher) {
+    public void initialize(ClusterConfigWatcher configWatcher) {
         this.configWatcher = configWatcher;
     }
 

src/main/java/org/geoserver/cluster/hazelcast/EventHzSynchronizer.java

@@ -1,5 +1,7 @@
 package org.geoserver.cluster.hazelcast;
 
+import static java.lang.String.format;
+
 import java.lang.reflect.Method;
 import java.lang.reflect.Proxy;
 import java.util.Iterator;
@@ -17,7 +19,6 @@
 import org.geoserver.catalog.StyleInfo;
 import org.geoserver.catalog.WorkspaceInfo;
 import org.geoserver.catalog.event.CatalogAddEvent;
-import org.geoserver.catalog.event.CatalogEvent;
 import org.geoserver.catalog.event.CatalogListener;
 import org.geoserver.catalog.event.CatalogPostModifyEvent;
 import org.geoserver.catalog.event.CatalogRemoveEvent;
@@ -35,6 +36,8 @@
 import org.geoserver.config.ServiceInfo;
 import org.geoserver.config.SettingsInfo;
 
+import com.google.common.collect.ImmutableList;
+
 /**
  * Synchronizer that converts cluster events and dispatches them the GeoServer config/catalog.
  * <p>
@@ -54,47 +57,30 @@ public EventHzSynchronizer(HzCluster cluster, GeoServer gs) {
     protected void processEventQueue(Queue<Event> q) throws Exception {
         Catalog cat = gs.getCatalog();
         Iterator<Event> it = q.iterator();
-        while (it.hasNext()) {
-            Event e = it.next();
-            if (e instanceof ConfigChangeEvent) {
-                ConfigChangeEvent ce = (ConfigChangeEvent) e;
+        while (it.hasNext() && isStarted()) {
+            final Event event = it.next();
+            it.remove();
+            LOGGER.fine(format("%s - Processing event %s", nodeId(), event));
+            if (event instanceof ConfigChangeEvent) {
+                ConfigChangeEvent ce = (ConfigChangeEvent) event;
                 Type t = ce.getChangeType();
                 Class<? extends Info> clazz = ce.getObjectInterface();
                 String id = ce.getObjectId();
                 String name = ce.getObjectName();
 
                 if (CatalogInfo.class.isAssignableFrom(clazz)) {
                     //catalog event
-                    CatalogInfo subj = null;
-                    if (WorkspaceInfo.class.isAssignableFrom(clazz)) {
-                        subj = cat.getWorkspace(id);
-                    }
-                    else if (NamespaceInfo.class.isAssignableFrom(clazz)) {
-                        subj = cat.getNamespace(id);
-                    }
-                    else if (StoreInfo.class.isAssignableFrom(clazz)) {
-                        subj = cat.getStore(id, (Class<StoreInfo>) clazz);
-                    }
-                    else if (ResourceInfo.class.isAssignableFrom(clazz)) {
-                        subj = cat.getResource(id, (Class<ResourceInfo>) clazz);
-                    }
-                    else if (LayerInfo.class.isAssignableFrom(clazz)) {
-                        subj = cat.getLayer(id);
-                    }
-                    else if (StyleInfo.class.isAssignableFrom(clazz)) {
-                        subj = cat.getStyle(id);
-                    }
-                    else if (LayerGroupInfo.class.isAssignableFrom(clazz)) {
-                        subj = cat.getLayerGroup(id);
-                    }
+                    CatalogInfo subj;
                     Method notifyMethod;
                     CatalogEventImpl evt;
                     switch(t) {
                     case ADD:
+                        subj = getCatalogInfo(cat, id, clazz);
                         notifyMethod = CatalogListener.class.getMethod("handleAddEvent", CatalogAddEvent.class);
                         evt = new CatalogAddEventImpl();
                         break;
                     case MODIFY:
+                        subj = getCatalogInfo(cat, id, clazz);
                         notifyMethod = CatalogListener.class.getMethod("handlePostModifyEvent", CatalogPostModifyEvent.class);
                         evt = new CatalogPostModifyEventImpl();
                         break;
@@ -108,26 +94,37 @@ else if (LayerGroupInfo.class.isAssignableFrom(clazz)) {
                         throw new IllegalStateException("Should not happen");
                     }
 
-                    if (subj == null) {
+                    if (subj == null) {//can't happen if type == DELETE
+                        ConfigChangeEvent removeEvent = new ConfigChangeEvent(id, name, clazz,
+                                Type.REMOVE);
+                        if (queue.contains(removeEvent)) {
+                            LOGGER.fine(format("%s - Ignoring event %s, a remove is queued.",
+                                    nodeId(), event));
+                            continue;
+                        }
                         //this could be latency in the catalog itself, abort processing since
                         // events need to processed in order and further events might depend 
                         // on this event
-                        LOGGER.warning(String.format("Received %s event for (%s, %s) but could" 
-                            + " not find in catalog", t.name(), id, clazz.getSimpleName()));
-                        return;
+                        String message = format(
+                                "%s - Error processing event %s but object not found in catalog", nodeId(),
+                                event);
+                        LOGGER.warning(message);
+                        continue;
                     }
 
                     evt.setSource(subj);
-
                     try {
-                        for (CatalogListener l:cat.getListeners()){
+                        for (CatalogListener l: ImmutableList.copyOf(cat.getListeners())){
                             // Don't notify self otherwise the event bounces back out into the
                             // cluster.
-                            if(l!=this) notifyMethod.invoke(l, evt);
+                            if (l != this && isStarted()) {
+                                notifyMethod.invoke(l, evt);
+                            }
                         }
                     }
                     catch(Exception ex) {
-                        LOGGER.log(Level.WARNING, "Event dispatch failed", ex);
+                        LOGGER.log(Level.WARNING,
+                                format("%s - Event dispatch failed: %s", nodeId(), event), ex);
                     }
 
                 }
@@ -158,13 +155,32 @@ else if (LayerGroupInfo.class.isAssignableFrom(clazz)) {
                             if(l!=this) notifyMethod.invoke(l, subj);
                         }
                         catch(Exception ex) {
-                            LOGGER.log(Level.WARNING, "Event dispatch failed", ex);
+                            LOGGER.log(Level.WARNING,
+                                    format("%s - Event dispatch failed: %s", nodeId(), event), ex);
                         }
                     }
                 }
             }
-
-            it.remove();
         }
     }
+
+    private CatalogInfo getCatalogInfo(Catalog cat, String id, Class<? extends Info> clazz) {
+        CatalogInfo subj = null;
+        if (WorkspaceInfo.class.isAssignableFrom(clazz)) {
+            subj = cat.getWorkspace(id);
+        } else if (NamespaceInfo.class.isAssignableFrom(clazz)) {
+            subj = cat.getNamespace(id);
+        } else if (StoreInfo.class.isAssignableFrom(clazz)) {
+            subj = cat.getStore(id, (Class<StoreInfo>) clazz);
+        } else if (ResourceInfo.class.isAssignableFrom(clazz)) {
+            subj = cat.getResource(id, (Class<ResourceInfo>) clazz);
+        } else if (LayerInfo.class.isAssignableFrom(clazz)) {
+            subj = cat.getLayer(id);
+        } else if (StyleInfo.class.isAssignableFrom(clazz)) {
+            subj = cat.getStyle(id);
+        } else if (LayerGroupInfo.class.isAssignableFrom(clazz)) {
+            subj = cat.getLayerGroup(id);
+        }
+        return subj;
+    }
 }