Change source instance id type from UUID to String

From Gerrit v3.2 event contains instance id field. In version 3.4
EventMessage envelope will be replaced with Event. To allow rolling
upgrade compatibility between v3.3 and v3.4 need to be assured. To do
that EventMessage.sourceInstanceId field type is changed from UUID to
String.

Also do not process events with sourceInstanceId equals null.

Bug: Issue 14390
Change-Id: I69aa4a3637fac99a7221fa8047d84125d4f4f6bd
diff --git a/external_plugin_deps.bzl b/external_plugin_deps.bzl
index 82cd5e8..223778e 100644
--- a/external_plugin_deps.bzl
+++ b/external_plugin_deps.bzl
@@ -9,6 +9,6 @@
 
     maven_jar(
         name = "events-broker",
-        artifact = "com.gerritforge:events-broker:3.3.1",
-        sha1 = "90775e671946b20e52be3a11277d1ed33973d66e",
+        artifact = "com.gerritforge:events-broker:3.3.2",
+        sha1 = "d8bcb77047cc12dd7c623b5b4de70a25499d3d6c",
     )
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/AbstractSubcriber.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/AbstractSubcriber.java
index 1f8766a..b37f434 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/AbstractSubcriber.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/AbstractSubcriber.java
@@ -15,6 +15,7 @@
 package com.googlesource.gerrit.plugins.multisite.consumer;
 
 import com.gerritforge.gerrit.eventbroker.EventMessage;
+import com.google.common.base.Strings;
 import com.google.common.flogger.FluentLogger;
 import com.google.gerrit.extensions.registration.DynamicSet;
 import com.google.gerrit.server.permissions.PermissionBackendException;
@@ -34,7 +35,7 @@
 
   private final ForwardedEventRouter eventRouter;
   private final DynamicSet<DroppedEventListener> droppedEventListeners;
-  private final UUID instanceId;
+  private final String instanceId;
   private final MessageLogger msgLog;
   private SubscriberMetrics subscriberMetrics;
   private final Configuration cfg;
@@ -49,7 +50,7 @@
       Configuration cfg) {
     this.eventRouter = eventRouter;
     this.droppedEventListeners = droppedEventListeners;
-    this.instanceId = instanceId;
+    this.instanceId = instanceId.toString();
     this.msgLog = msgLog;
     this.subscriberMetrics = subscriberMetrics;
     this.cfg = cfg;
@@ -63,11 +64,18 @@
   }
 
   private void processRecord(EventMessage event) {
+    String sourceInstanceId = event.getHeader().sourceInstanceId;
 
-    if (event.getHeader().sourceInstanceId.equals(instanceId)) {
-      logger.atFiner().log(
-          "Dropping event %s produced by our instanceId %s",
-          event.toString(), instanceId.toString());
+    if (Strings.isNullOrEmpty(sourceInstanceId) || instanceId.equals(sourceInstanceId)) {
+      if (Strings.isNullOrEmpty(sourceInstanceId)) {
+        logger.atWarning().log(
+            String.format(
+                "Dropping event %s because sourceInstanceId cannot be null", event.toString()));
+      } else {
+        logger.atFiner().log(
+            String.format(
+                "Dropping event %s produced by our instanceId %s", event.toString(), instanceId));
+      }
       droppedEventListeners.forEach(l -> l.onEventDropped(event));
     } else {
       try {