Fixed hz-cluster unit test failures. Upgraded Hazelcast from 3.11-BET…

…A-1 to 3.11. Upgraded Metrics from 2.1.5 to 4.0.3.

src/community/hz-cluster/pom.xml

@@ -38,7 +38,7 @@
       <version>3.3.1</version>
     </dependency>
     <dependency>
-      <groupId>com.yammer.metrics</groupId>
+      <groupId>io.dropwizard.metrics</groupId>
       <artifactId>metrics-servlet</artifactId>
       <version>${metrics.version}</version>
     </dependency>
@@ -100,7 +100,7 @@
   </dependencies>
 
   <properties>
-    <metrics.version>2.1.5</metrics.version>
+    <metrics.version>4.0.3</metrics.version>
   </properties>
 </project>
 

src/community/hz-cluster/src/main/java/org/geoserver/cluster/ConfigChangeEvent.java

@@ -26,6 +26,8 @@
 import org.geoserver.catalog.StyleInfo;
 import org.geoserver.catalog.WMSLayerInfo;
 import org.geoserver.catalog.WMSStoreInfo;
+import org.geoserver.catalog.WMTSLayerInfo;
+import org.geoserver.catalog.WMTSStoreInfo;
 import org.geoserver.catalog.WorkspaceInfo;
 import org.geoserver.catalog.impl.AttributionInfoImpl;
 import org.geoserver.catalog.impl.CatalogImpl;
@@ -40,6 +42,8 @@
 import org.geoserver.catalog.impl.StyleInfoImpl;
 import org.geoserver.catalog.impl.WMSLayerInfoImpl;
 import org.geoserver.catalog.impl.WMSStoreInfoImpl;
+import org.geoserver.catalog.impl.WMTSLayerInfoImpl;
+import org.geoserver.catalog.impl.WMTSStoreInfoImpl;
 import org.geoserver.catalog.impl.WorkspaceInfoImpl;
 import org.geoserver.config.ContactInfo;
 import org.geoserver.config.GeoServerInfo;
@@ -74,11 +78,13 @@ public class ConfigChangeEvent extends Event {
         INTERFACES.put(WorkspaceInfoImpl.class, WorkspaceInfo.class);
         INTERFACES.put(DataStoreInfoImpl.class, DataStoreInfo.class);
         INTERFACES.put(WMSStoreInfoImpl.class, WMSStoreInfo.class);
+        INTERFACES.put(WMTSStoreInfoImpl.class, WMTSStoreInfo.class);
         INTERFACES.put(CoverageStoreInfoImpl.class, CoverageStoreInfo.class);
         INTERFACES.put(StyleInfoImpl.class, StyleInfo.class);
         INTERFACES.put(FeatureTypeInfoImpl.class, FeatureTypeInfo.class);
         INTERFACES.put(CoverageInfoImpl.class, CoverageInfo.class);
         INTERFACES.put(WMSLayerInfoImpl.class, WMSLayerInfo.class);
+        INTERFACES.put(WMTSLayerInfoImpl.class, WMTSLayerInfo.class);
         INTERFACES.put(MetadataLinkInfoImpl.class, MetadataLinkInfo.class);
         INTERFACES.put(LayerInfoImpl.class, LayerInfo.class);
         INTERFACES.put(LayerGroupInfoImpl.class, LayerGroupInfo.class);

src/community/hz-cluster/src/main/java/org/geoserver/cluster/hazelcast/EventHzSynchronizer.java

@@ -16,13 +16,13 @@
 import com.hazelcast.core.Member;
 import com.hazelcast.core.Message;
 import com.hazelcast.core.MessageListener;
-import com.yammer.metrics.Metrics;
 import java.lang.reflect.Method;
 import java.lang.reflect.Proxy;
 import java.net.InetSocketAddress;
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.Future;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.logging.Level;
 import javax.annotation.Nullable;
@@ -93,7 +93,7 @@ protected void dispatch(Event e) {
         e.setSource(localAddress(cluster.getHz()));
         topic.publish(e);
 
-        Metrics.newCounter(getClass(), "dispatched").inc();
+        incCounter(getClass(), "dispatched");
         waitForAck(e);
     }
 
@@ -157,10 +157,10 @@ private void waitForAck(Event event) {
     }
 
     @Override
-    protected void processEvent(Event event) throws Exception {
+    protected Future<?> processEvent(Event event) {
         Preconditions.checkState(isStarted());
         if (!(event instanceof ConfigChangeEvent)) {
-            return;
+            return null;
         }
         try {
             LOGGER.fine(format("%s - Processing event %s", nodeId(), event));
@@ -176,6 +176,7 @@ protected void processEvent(Event event) throws Exception {
         } finally {
             ack(event);
         }
+        return null;
     }
 
     private void processCatalogEvent(final ConfigChangeEvent event)

src/community/hz-cluster/src/main/java/org/geoserver/cluster/hazelcast/HzSynchronizer.java

@@ -8,14 +8,16 @@
 import static java.lang.String.format;
 import static org.geoserver.cluster.hazelcast.HazelcastUtil.localAddress;
 
+import com.codahale.metrics.MetricRegistry;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import com.hazelcast.core.ITopic;
 import com.hazelcast.core.Message;
 import com.hazelcast.core.MessageListener;
-import com.yammer.metrics.Metrics;
 import java.util.List;
 import java.util.Queue;
+import java.util.concurrent.Callable;
 import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.logging.Level;
@@ -56,6 +58,8 @@ public abstract class HzSynchronizer extends GeoServerSynchronizer
 
     protected static Logger LOGGER = Logging.getLogger("org.geoserver.cluster.hazelcast");
 
+    private final MetricRegistry registry = new MetricRegistry();
+
     protected final HzCluster cluster;
 
     protected final ITopic<Event> topic;
@@ -96,7 +100,7 @@ public void onMessage(Message<Event> message) {
             }
             return;
         }
-        Metrics.newCounter(getClass(), "recieved").inc();
+        incCounter(getClass(), "recieved");
         if (localAddress(cluster.getHz()).equals(event.getSource())) {
             if (LOGGER.isLoggable(Level.FINER)) {
                 LOGGER.finer(
@@ -113,7 +117,7 @@ public void onMessage(Message<Event> message) {
         executor.schedule(new EventWorker(event), syncDelay, TimeUnit.SECONDS);
     }
 
-    private class EventWorker implements Runnable {
+    private class EventWorker implements Callable<Future<?>> {
 
         private Event event;
 
@@ -122,17 +126,19 @@ public EventWorker(Event event) {
         }
 
         @Override
-        public void run() {
+        public Future<?> call() {
             if (!isStarted()) {
-                return;
+                return null;
             }
+            Future<?> future = null;
             try {
-                processEvent(event);
+                future = processEvent(event);
             } catch (Exception e) {
                 LOGGER.log(Level.WARNING, format("%s - Event processing failed", nodeId()), e);
             }
 
-            Metrics.newCounter(getClass(), "reloads").inc();
+            incCounter(getClass(), "reloads");
+            return future;
         }
     }
 
@@ -144,7 +150,7 @@ public void run() {
      * <p><b>Note:</b> It is the responsibility of subclasses to clear events from the queue as they
      * are processed.
      */
-    protected abstract void processEvent(Event event) throws Exception;
+    protected abstract Future<?> processEvent(Event event);
 
     ConfigChangeEvent newChangeEvent(CatalogEvent evt, Type type) {
         return newChangeEvent(evt.getSource(), type);
@@ -225,6 +231,11 @@ public void handleSettingsRemoved(SettingsInfo settings) {
         dispatch(newChangeEvent(settings, Type.REMOVE));
     }
 
+    /** Increments the counter for the specified class and name by one. */
+    protected void incCounter(Class<?> clazz, String name) {
+        this.registry.counter(MetricRegistry.name(clazz, name)).inc();
+    }
+
     protected String nodeId() {
         return HazelcastUtil.nodeId(cluster);
     }

src/community/hz-cluster/src/main/java/org/geoserver/cluster/hazelcast/ReloadHzSynchronizer.java

@@ -9,10 +9,11 @@
 import static org.geoserver.cluster.hazelcast.HazelcastUtil.localAddress;
 
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
-import com.yammer.metrics.Metrics;
+import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
 import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.RejectedExecutionHandler;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
@@ -43,28 +44,24 @@ public ReloadHzSynchronizer(HzCluster cluster, GeoServer gs) {
                         .setDaemon(true)
                         .setNameFormat("Hz-GeoServer-Reload-%d")
                         .build();
-        RejectedExecutionHandler rejectionHandler = new ThreadPoolExecutor.DiscardPolicy();
         // a thread pool executor operating out of a blocking queue with maximum of 1 element, which
         // discards execute requests if the queue is full
         reloadService =
                 new ThreadPoolExecutor(
-                        1,
-                        1,
-                        0L,
-                        TimeUnit.MILLISECONDS,
-                        new LinkedBlockingQueue<Runnable>(1),
-                        threadFactory,
-                        rejectionHandler);
+                        1, 1, 0L, TimeUnit.MILLISECONDS, getWorkQueue(), threadFactory);
+    }
+
+    BlockingQueue<Runnable> getWorkQueue() {
+        return new LinkedBlockingQueue<>(1);
     }
 
     @Override
-    protected void processEvent(Event event) throws Exception {
+    protected Future<?> processEvent(Event event) {
         // submit task and return immediately. The task will be ignored if another one is already
         // scheduled
-        reloadService.submit(
-                new Runnable() {
-                    @Override
-                    public void run() {
+        try {
+            return reloadService.submit(
+                    () -> {
                         // lock during event processing
                         eventLock.set(true);
                         try {
@@ -74,8 +71,14 @@ public void run() {
                         } finally {
                             eventLock.set(false);
                         }
-                    }
-                });
+                    });
+        } catch (RejectedExecutionException e) {
+            if (LOGGER.isLoggable(Level.FINEST)) {
+                LOGGER.finest(
+                        format("%s - Reload in progress. Ignoring event %s", nodeId(), event));
+            }
+            return null;
+        }
     }
 
     @Override
@@ -90,6 +93,6 @@ protected void dispatch(Event e) {
         e.setSource(localAddress(cluster.getHz()));
         topic.publish(e);
 
-        Metrics.newCounter(getClass(), "dispatched").inc();
+        incCounter(getClass(), "dispatched");
     }
 }

src/community/hz-cluster/src/main/java/org/geoserver/cluster/hazelcast/RemovedObjectProxy.java

@@ -30,6 +30,8 @@
 import org.geoserver.catalog.StyleInfo;
 import org.geoserver.catalog.WMSLayerInfo;
 import org.geoserver.catalog.WMSStoreInfo;
+import org.geoserver.catalog.WMTSLayerInfo;
+import org.geoserver.catalog.WMTSStoreInfo;
 import org.geoserver.catalog.WorkspaceInfo;
 import org.geotools.data.DataStore;
 
@@ -124,12 +126,16 @@ private void proxyVisitory(Object proxy, Method method, CatalogVisitor catalogVi
             catalogVisitor.visit((FeatureTypeInfo) proxy);
         } else if (WMSLayerInfo.class.equals(infoInterface)) {
             catalogVisitor.visit((WMSLayerInfo) proxy);
+        } else if (WMTSLayerInfo.class.equals(infoInterface)) {
+            catalogVisitor.visit((WMTSLayerInfo) proxy);
         } else if (CoverageStoreInfo.class.equals(infoInterface)) {
             catalogVisitor.visit((CoverageStoreInfo) proxy);
         } else if (DataStoreInfo.class.equals(infoInterface)) {
             catalogVisitor.visit((DataStoreInfo) proxy);
         } else if (WMSStoreInfo.class.equals(infoInterface)) {
             catalogVisitor.visit((WMSStoreInfo) proxy);
+        } else if (WMTSStoreInfo.class.equals(infoInterface)) {
+            catalogVisitor.visit((WMTSStoreInfo) proxy);
         } else if (StyleInfo.class.equals(infoInterface)) {
             catalogVisitor.visit((StyleInfo) proxy);
         } else if (LayerInfo.class.equals(infoInterface)) {

src/community/hz-cluster/src/test/java/org/geoserver/cluster/hazelcast/HzSynchronizerTest.java

@@ -26,10 +26,13 @@
 import java.io.IOException;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Iterator;
 import java.util.List;
 import java.util.UUID;
+import java.util.concurrent.Callable;
+import java.util.concurrent.Future;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import org.apache.commons.io.FileUtils;
@@ -162,7 +165,7 @@ public Object answer() throws Throwable {
         expectLastCall().atLeastOnce();
 
         executor = createMock(ScheduledExecutorService.class);
-        captureExecutor = new Capture<Runnable>(CaptureType.ALL);
+        captureExecutor = new Capture<>(CaptureType.ALL);
         expect(executor.schedule(capture(captureExecutor), anyLong(), (TimeUnit) anyObject()))
                 .andStubReturn(null);
     }
@@ -192,7 +195,7 @@ MessageListener<Event> getListener() {
     protected Capture<CatalogListener> catListenerCapture;
     protected Capture<MessageListener<Event>> captureTopicListener;
     protected Capture<MessageListener<UUID>> captureAckTopicListener;
-    protected Capture<Runnable> captureExecutor;
+    protected Capture<Callable<Future<?>>> captureExecutor;
     protected Capture<UUID> captureAckTopicPublish;
 
     public List<Object> myMocks() {
@@ -267,12 +270,19 @@ protected void waitForSync() throws Exception {
         // Thread.sleep(SYNC_DELAY*1000+500); // Convert to millis, then add a little extra to be
         // sure
 
-        List<Runnable> tasks = captureExecutor.getValues();
+        List<Callable<Future<?>>> tasks = captureExecutor.getValues();
+        List<Future<?>> futures = new ArrayList<>(1);
 
-        for (Iterator<Runnable> i = tasks.iterator(); i.hasNext(); ) {
-            Runnable task = i.next();
+        for (Iterator<Callable<Future<?>>> i = tasks.iterator(); i.hasNext(); ) {
+            Callable<Future<?>> task = i.next();
             i.remove();
-            task.run();
+            Future<?> future = task.call();
+            if (future != null) {
+                futures.add(future);
+            }
+        }
+        for (Future<?> future : futures) {
+            future.get();
         }
     }