Hazelcast impl

.gitignore

@@ -19,3 +19,4 @@ hostkey.ser
 /eclipse-classes
 .vscode/
 .factorypath
+graphify-out

README.md

@@ -14,7 +14,6 @@ Each job can be configured with one Gerrit server.
 ## Maintainers
 
 * Robert Sandell
-  - robert.sandell@cloudbees.com
   - sandell.robert@gmail.com
 
 * Tomas Westling
@@ -52,6 +51,13 @@ Run checkstyle
 
     mvn checkstyle:checkstyle
 
+# Distributed Event Management support
+
+The plugin supports an distributed event management for the memory that will track the events (for example using Hazelcast client).
+See [README_DISTRIBUTED_EVENT_MANAGEMENT.md](docs/README_DISTRIBUTED_EVENT_MANAGEMENT.md) for configuration
+properties and deployment examples.
+
+
 # License
 
     The MIT License

docs/README_DISTRIBUTED_EVENT_MANAGEMENT.md

@@ -0,0 +1,107 @@
+# Distributed Event Management support
+
+The plugin supports Distributed Event Management support where two or more replicas or nodes of a logical Jenkins(*) instance
+run in parallel (sharing the Gerrit memory of the plugin). When enabled, a Hazelcast
+cluster coordinates the instances so that:
+
+- Each Gerrit event is processed by **exactly one** instance (event claiming)
+- Build state is shared across instances (distributed build memory)
+- Gerrit feedback (votes and comments) are sent **exactly once** per build event
+
+By default, the plugin runs in **local mode** and requires no additional configuration.
+Local mode is fully backward-compatible with single-instance Jenkins deployments.
+
+Alternative coordination backends can be implemented by extending
+[`CoordinationModeProvider`](../src/main/java/com/sonyericsson/hudson/plugins/gerrit/trigger/spi/CoordinationModeProvider.java)
+— a Jenkins `ExtensionPoint` that wires together the storage, event-claiming, and
+notification-claiming strategies for a given coordination mode. A higher `@Extension`
+ordinal takes precedence over the built-in Hazelcast provider.
+
+## Hazelcast implementation
+
+Hazelcast mode is activated via a JVM system property. Jenkins connects as a lightweight
+client to a Hazelcast sidecar container, reusing the cross-pod cluster the sidecar
+maintains.
+
+### Configuration Properties
+
+All distributed storage settings are controlled by JVM system properties passed to Jenkins on startup.
+
+| Property | Default | Description                                           |
+|---|---|-------------------------------------------------------|
+| `gerrit.trigger.coordination.mode` | `local` | Set to `hazelcast` to enable distributed coordination |
+| `gerrit.trigger.coordination.hazelcast.client.addresses` | `localhost:5702` | Comma-separated `host:port` list of sidecar addresses |
+| `gerrit.trigger.coordination.hazelcast.client.cluster.name` | `gerrit-trigger-cluster` | Cluster name to connect to                            |
+
+Port `5702` is used by default to avoid potential conflicts with other Hazelcast cluster, which could occupy port `5701`.
+
+Cluster name must be different for each logical instance. Multiple replicas or nodes of a logical instance may configure the same cluster name. Different logical instances require separate cluster names.
+
+### Configuration Example
+
+#### Kubernetes — Client Mode with Hazelcast Sidecar
+
+The plugin can connect to Hazelcast cluster as a lightweight client. For example, if we are running 
+K8s environment with the Jenkins instance inside a pod, we can have a side-container with Hazelcast
+to set up the Hazelcast cluster. In this kind of cases, we would the a configuration setup similar
+to the following one:
+
+Add the following JVM arguments to the Jenkins instance:
+
+    -Dgerrit.trigger.coordination.mode=hazelcast
+    -Dgerrit.trigger.coordination.hazelcast.client.addresses=localhost:5702
+    -Dgerrit.trigger.coordination.hazelcast.client.cluster.name=gerrit-trigger-cluster
+
+Add the sidecar container to the instance pod spec:
+
+```yaml
+- name: hazelcast
+  image: hazelcast/hazelcast:5.3.8
+  ports:
+    - containerPort: 5702
+      name: hazelcast
+  env:
+    - name: JAVA_OPTS
+      value: >-
+        -Dhazelcast.config=/dev/stdin
+        -Dhazelcast.local.publicAddress=$(POD_IP):5702
+    - name: HZ_CLUSTERNAME
+      value: gerrit-trigger-cluster
+    - name: HZ_NETWORK_PORT_PORT
+      value: "5702"
+```
+
+Grant the pod's service account read access to Kubernetes endpoints so Hazelcast can
+discover its peers:
+
+```yaml
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRole
+metadata:
+  name: hazelcast-gerrit-trigger
+rules:
+  - apiGroups: [""]
+    resources: ["endpoints", "pods", "nodes", "services"]
+    verbs: ["get", "list"]
+  - apiGroups: ["discovery.k8s.io"]
+    resources: ["endpointslices"]
+    verbs: ["get", "list"]
+```
+
+#### Kubernetes — Client Mode with Separate Hazelcast Cluster
+
+For larger deployments or strict separation of concerns, you can decouple the coordination layer by running a standalone Hazelcast cluster. Jenkins still connects as a lightweight client, but routes traffic to the separate cluster via a Kubernetes service instead of a sidecar.
+
+Add the following JVM arguments to the Jenkins instance, updating the client address to point to your standalone Hazelcast Kubernetes service (replace hazelcast-service.default.svc.cluster.local with your actual service DNS and namespace, along with the cluster name for your logical instance):
+
+    -Dgerrit.trigger.coordination.mode=hazelcast
+    -Dgerrit.trigger.coordination.hazelcast.client.addresses=hazelcast-service.default.svc.cluster.local:5702
+    -Dgerrit.trigger.coordination.hazelcast.client.cluster.name=gerrit-trigger-cluster-<LOGICAL_INSTANCE_NAME>
+
+In this topology:
+- You do not need to add the sidecar container to the Jenkins pod spec.
+- The Jenkins service account does not need RBAC permissions for peer discovery, as cluster management is handled entirely by the standalone Hazelcast nodes.
+- You must deploy and manage the Hazelcast cluster independently (e.g. via the official Hazelcast Helm chart), ensuring you configure it to match your expected `HZ_CLUSTERNAME` and port (`5702`).
+
+(*) Jenkins does not support multiple replicas or nodes for a single logical instance, this feature is not tested with Jenkins. This feature is provided for CloudBees CI (Enterprise Jenkins).
+This feature is provided as a community effort and is not endorsed or officially supported by CloudBees.

pom.xml

@@ -67,6 +67,7 @@
         <surefire.rerunFailingTestsCount>3</surefire.rerunFailingTestsCount>
         <forkCount>0.5C</forkCount>
         <spotbugs.threshold>High</spotbugs.threshold>
+        <hazelcast.version>5.3.8</hazelcast.version>
     </properties>
 
     <dependencies>
@@ -107,6 +108,11 @@
             </exclusions>
             <!-- New source is here: https://github.com/sonyxperiadev/gerrit-events -->
         </dependency>
+        <dependency>
+            <groupId>com.hazelcast</groupId>
+            <artifactId>hazelcast</artifactId>
+            <version>${hazelcast.version}</version>
+        </dependency>
         <dependency>
             <groupId>io.jenkins.plugins</groupId>
             <artifactId>gson-api</artifactId>
@@ -203,6 +209,12 @@
             <artifactId>workflow-support</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.jenkins-ci.plugins.workflow</groupId>
+            <artifactId>workflow-support</artifactId>
+            <classifier>tests</classifier>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <!-- Used for test with matrix permissions -->
             <groupId>org.jenkins-ci.plugins</groupId>
@@ -370,6 +382,42 @@
         </plugins>
     </build>
 
+    <profiles>
+        <!--
+            Profile for testing with Hazelcast coordination mode enabled.
+            Usage: mvn clean test -Ptest-hazelcast
+
+            This profile enables Hazelcast distributed coordination during tests to verify:
+            - HazelcastCoordinationProvider is selected
+            - HazelcastBuildMemoryStorage works correctly
+            - Event and notification claiming strategies work
+
+            An embedded Hazelcast server is automatically started before any tests run by
+            HazelcastServerTestListener (discovered via
+            META-INF/services/org.junit.platform.launcher.TestExecutionListener). This provides
+            the localhost:5702 endpoint that the plugin client connects to, with no external
+            Kubernetes or TCP infrastructure required.
+        -->
+        <profile>
+            <id>test-hazelcast</id>
+            <build>
+                <plugins>
+                    <plugin>
+                        <artifactId>maven-surefire-plugin</artifactId>
+                        <configuration>
+                            <reuseForks>false</reuseForks>
+                            <forkCount>1</forkCount>
+                            <systemPropertyVariables>
+                                <!-- Enable Hazelcast coordination mode for tests -->
+                                <gerrit.trigger.coordination.mode>hazelcast</gerrit.trigger.coordination.mode>
+                            </systemPropertyVariables>
+                        </configuration>
+                    </plugin>
+                </plugins>
+            </build>
+        </profile>
+    </profiles>
+
     <scm>
         <connection>scm:git:https://github.com/${gitHubRepo}.git</connection>
         <developerConnection>scm:git:git@github.com:${gitHubRepo}.git</developerConnection>

src/main/java/com/sonyericsson/hudson/plugins/gerrit/trigger/PluginImpl.java

@@ -24,8 +24,11 @@
  */
 package com.sonyericsson.hudson.plugins.gerrit.trigger;
 
+import com.sonyericsson.hudson.plugins.gerrit.trigger.coordination.CoordinationModeFactory;
+import com.sonyericsson.hudson.plugins.gerrit.trigger.coordination.LocalCoordinationProvider;
 import com.sonyericsson.hudson.plugins.gerrit.trigger.dependency.DependencyQueueTaskDispatcher;
 import com.sonyericsson.hudson.plugins.gerrit.trigger.replication.ReplicationQueueTaskDispatcher;
+import com.sonyericsson.hudson.plugins.gerrit.trigger.spi.CoordinationModeProvider;
 import com.sonymobile.tools.gerrit.gerritevents.GerritHandler;
 import com.sonymobile.tools.gerrit.gerritevents.GerritSendCommandQueue;
 import com.sonyericsson.hudson.plugins.gerrit.trigger.config.Config;
@@ -583,6 +586,18 @@
         logger.info("Starting Gerrit-Trigger Plugin");
         logger.trace("Loading configs");
         load();
+
+        // Initialize coordination providers early (before any code that might use CoordinationModeFactory)
+        // This must happen before BuildMemory, EventClaimStrategy, or NotificationClaimStrategy are used
+        // because provider.isAvailable() may check if resources are initialized
+        initializeCoordinationProviders();
+
+        // Eagerly initialize CoordinationModeFactory so discoverMode() runs now (during startup)
+        // rather than lazily on first event — deferred initialization can add several seconds of
+        // latency to the first build trigger when ExtensionList.lookup() is called from a
+        // background event-processing thread.
+        CoordinationModeFactory.get().getStorage();
+
         GerritSendCommandQueue.initialize(pluginConfig);
         gerritEventManager = new JenkinsAwareGerritHandler(pluginConfig.getNumberOfReceivingWorkerThreads());
         for (GerritServer s : servers) {
@@ -591,101 +606,194 @@
         active = true;
     }
 
+    /**
+     * Initialize the active coordination mode provider.
+     * <p>
+     * This is called early in plugin startup, before any code that might use
+     * CoordinationModeFactory. Only initializes the provider that matches the configured
+     * coordination mode, making it more efficient than calling initialize() on all providers.
+     * <p>
+     * <strong>Implementation Note:</strong> We cannot use {@code provider.isAvailable()}
+     * before initialization because isAvailable() checks if the provider is actually initialized.
+     * Instead, we check the configured mode directly and initialize the matching provider.
+     * After initialization, isAvailable() will return true.
+     * <p>
+     * Fails gracefully - if a provider's initialization fails, it will not be available
+     * and the factory will fall back to the next highest-priority provider.
+     */
+    private void initializeCoordinationProviders() {
+        ExtensionList<CoordinationModeProvider> providers = ExtensionList.lookup(CoordinationModeProvider.class);
+        String configuredMode = CoordinationModeProvider.getConfiguredMode();
+        logger.debug("Configured coordination mode: {}", configuredMode);
+
+        // Try to initialize the configured (non-local) provider first.
+        // Each provider's initialize() internally checks the configured mode and is a no-op
+        // if the mode doesn't match, so we try non-local providers in priority order.
+        // Local is handled separately below as the explicit fallback.
+        if (!"local".equalsIgnoreCase(configuredMode)) {
+            for (CoordinationModeProvider provider : providers) {
+                if (provider instanceof LocalCoordinationProvider) {
+                    continue;
+                }
+                try {
+                    logger.info("Initializing coordination provider: {}", provider.getModeName());
+                    provider.initialize();
+                    if (provider.isAvailable()) {
+                        logger.info("Provider {} initialized successfully", provider.getModeName());
+                        return;
+                    }
+                } catch (Exception e) {
+                    logger.warn("Failed to initialize {} coordination provider. Falling back to Local.",
+                            provider.getModeName(), e);
+                    break;
+                }
+            }
+        }
+
+        // Explicit fallback to LocalCoordinationProvider, which is always available.
+        for (CoordinationModeProvider provider : providers) {
+            if (provider instanceof LocalCoordinationProvider) {
+                if ("local".equalsIgnoreCase(configuredMode)) {
+                    logger.info("Initializing LocalCoordinationProvider");
+                } else {
+                    logger.info("Initializing LocalCoordinationProvider (fallback after failed initialization)");
+                }
+                try {
+                    provider.initialize();
+                } catch (Exception e) {
+                    logger.error("Failed to initialize LocalCoordinationProvider - this should never happen", e);
+                }
+                return;
+            }
+        }
+
+        logger.error("LocalCoordinationProvider not found - this should never happen");
+    }
+
     /**
      * Forces initialization of the Dispatchers.
      *
      * They need to register and listen to GerritEvents. Normally, it is lazy loaded when the first build is started.
      */
     @Initializer(after = InitMilestone.SYSTEM_CONFIG_ADAPTED)
     public static void initializeDispatchers() {
         // Call the following method for force initialization of the Dispatchers because
         // it needs to register and listen to GerritEvent. Normally, it is lazy loaded when the first build is started.
         try {
             ExtensionList.lookupSingleton(ReplicationQueueTaskDispatcher.class);
         } catch (IllegalStateException e) {
             logger.warn("Failed to initialize Replication Queue.", e);
         }
         try {
             ExtensionList.lookupSingleton(DependencyQueueTaskDispatcher.class);
         } catch (IllegalStateException e) {
             logger.warn("Failed to initialize Dependency Queue.", e);
         }
     }
 
     /**
      * Load plugin config.
      */
     @Override
     public void load() {
         super.load();
         if (pluginConfig == null) {
             PluginConfig conf = new PluginConfig();
             if (config != null) {
                 conf.setNumberOfReceivingWorkerThreads(config.getNumberOfReceivingWorkerThreads());
                 conf.setNumberOfSendingWorkerThreads(config.getNumberOfSendingWorkerThreads());
             }
             pluginConfig = conf;
         }
         if (servers.isEmpty()) {
             if (config != null) { //have loaded data in old format, so add a new server with the old config to the list.
                 GerritServer defaultServer = new GerritServer(DEFAULT_SERVER_NAME);
                 defaultServer.setConfig(config);
                 servers.add(defaultServer);
             }
             save();
         }
         pluginConfig.updateEventFilter();
         //For unit/integration testing only...
         if (System.getProperty(TEST_SSH_KEYFILE_LOCATION_PROPERTY) != null && !servers.isEmpty()) {
             File location = new File(System.getProperty(TEST_SSH_KEYFILE_LOCATION_PROPERTY));
             for (GerritServer server : servers) {
                 ((Config)server.getConfig()).setGerritAuthKeyFile(location);
             }
         }
     }
 
     /**
      * Registers XStream alias and converters to handle backwards compatibility with old data.
      */
     protected static void doXStreamRegistrations() {
         logger.trace("doing XStream alias registrations.");
 
         //Register it in all known XStreams just to be sure.
         Items.XSTREAM.registerConverter(new TriggerContextConverter());
         Jenkins.XSTREAM.registerConverter(new TriggerContextConverter());
         //This is where the problems where, reading builds.
         Run.XSTREAM.registerConverter(new TriggerContextConverter());
         Run.XSTREAM2.addCompatibilityAlias(
             "com.sonyericsson.hudson.plugins.gerrit.gerritevents.dto.events.ManualPatchsetCreated",
             ManualPatchsetCreated.class);
 
         Items.XSTREAM.aliasPackage("com.sonyericsson.hudson.plugins.gerrit.gerritevents",
                 "com.sonymobile.tools.gerrit.gerritevents");
         Jenkins.XSTREAM.aliasPackage("com.sonyericsson.hudson.plugins.gerrit.gerritevents",
                 "com.sonymobile.tools.gerrit.gerritevents");
         Run.XSTREAM.aliasPackage("com.sonyericsson.hudson.plugins.gerrit.gerritevents",
                 "com.sonymobile.tools.gerrit.gerritevents");
 
         logger.trace("XStream alias registrations done.");
     }
 
     /**
      * Stop the plugin.
      */
     public void stop() {
         active = false;
+
+        // Shutdown coordination providers before stopping servers
+        // This ensures any coordination operations are cleaned up before servers disconnect
+        shutdownCoordinationProviders();
+
         for (GerritServer s : servers) {
             s.stop();
         }
         if (gerritEventManager != null) {
             gerritEventManager.shutdown(false);
             //TODO save to registered listeners?
             gerritEventManager = null;
         }
         GerritSendCommandQueue.shutdown();
         servers.clear();
     }
 
+    /**
+     * Shutdown all coordination mode providers.
+     * <p>
+     * Called during plugin shutdown to clean up coordination resources.
+     * Fails gracefully - errors are logged but don't prevent plugin shutdown.
+     */
+    private void shutdownCoordinationProviders() {
+        logger.debug("Shutting down coordination providers...");
+        for (com.sonyericsson.hudson.plugins.gerrit.trigger.spi.CoordinationModeProvider provider
+                : hudson.ExtensionList.lookup(
+                        com.sonyericsson.hudson.plugins.gerrit.trigger.spi.CoordinationModeProvider.class)) {
+            try {
+                logger.debug("Shutting down provider: {}", provider.getModeName());
+                provider.shutdown();
+                logger.debug("Provider {} shut down successfully", provider.getModeName());
+            } catch (Exception e) {
+                logger.warn("Error shutting down coordination provider: {} (non-critical, continuing shutdown)",
+                        provider.getModeName(), e);
+                // Continue with other providers even if one fails
+            }
+        }
+        logger.debug("Coordination provider shutdown complete");
+    }
+
     /**
      * Startup hook.
      */