Merge branch 'stable-3.3'

* stable-3.3:
  Deserialize Event and EventMessage
  Fix issue with message serialisation
  Bump events-broker to v3.3.2
  Add message content validation

Change-Id: I86d8e4f559ec4bf979ab31dded198abfbfc6553f

external_plugin_deps.bzl

diff --git a/external_plugin_deps.bzl b/external_plugin_deps.bzl
index 70bbdc8..d26ffb9 100644
--- a/external_plugin_deps.bzl
+++ b/external_plugin_deps.bzl
@@ -161,4 +161,4 @@
         name = "grpc-auth",
         artifact = "io.grpc:grpc-auth:1.36.0",
         sha1 = "d9722016658f8e649111c8bb93b299ea38dc207e",
-    )
\ No newline at end of file
+    )

src/main/java/com/googlesource/gerrit/plugins/pubsub/Module.java

diff --git a/src/main/java/com/googlesource/gerrit/plugins/pubsub/Module.java b/src/main/java/com/googlesource/gerrit/plugins/pubsub/Module.java
index 82e6133..32110dd 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/pubsub/Module.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/pubsub/Module.java
@@ -14,11 +14,11 @@
 
 package com.googlesource.gerrit.plugins.pubsub;
 
+import com.gerritforge.gerrit.eventbroker.EventGsonProvider;
 import com.google.api.gax.core.CredentialsProvider;
 import com.google.gerrit.extensions.config.FactoryModule;
 import com.google.gerrit.extensions.events.LifecycleListener;
 import com.google.gerrit.extensions.registration.DynamicSet;
-import com.google.gerrit.server.events.EventGsonProvider;
 import com.google.gerrit.server.events.EventListener;
 import com.google.gson.Gson;
 import com.google.inject.Inject;

src/main/java/com/googlesource/gerrit/plugins/pubsub/PubSubEventSubscriber.java

diff --git a/src/main/java/com/googlesource/gerrit/plugins/pubsub/PubSubEventSubscriber.java b/src/main/java/com/googlesource/gerrit/plugins/pubsub/PubSubEventSubscriber.java
index a6058ae..c7be04c 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/pubsub/PubSubEventSubscriber.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/pubsub/PubSubEventSubscriber.java
@@ -14,17 +14,22 @@
 
 package com.googlesource.gerrit.plugins.pubsub;
 
+import static java.util.Objects.requireNonNull;
+
 import com.gerritforge.gerrit.eventbroker.EventMessage;
+import com.gerritforge.gerrit.eventbroker.EventMessage.Header;
 import com.google.cloud.pubsub.v1.AckReplyConsumer;
 import com.google.cloud.pubsub.v1.MessageReceiver;
 import com.google.cloud.pubsub.v1.Subscriber;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.flogger.FluentLogger;
+import com.google.gerrit.server.events.Event;
 import com.google.gson.Gson;
 import com.google.inject.Inject;
 import com.google.inject.assistedinject.Assisted;
 import com.google.pubsub.v1.PubsubMessage;
 import java.io.IOException;
+import java.util.UUID;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.function.Consumer;
@@ -100,16 +105,34 @@
   MessageReceiver getMessageReceiver() {
     return (PubsubMessage message, AckReplyConsumer consumer) -> {
       try {
-        EventMessage event = gson.fromJson(message.getData().toStringUtf8(), EventMessage.class);
+        EventMessage event = deserialise(message.getData().toStringUtf8());
         messageProcessor.accept(event);
-        consumer.ack();
         subscriberMetrics.incrementSucceedToConsumeMessage();
       } catch (Exception e) {
         logger.atSevere().withCause(e).log(
             "Exception when consuming message %s from topic %s [message: %s]",
             message.getMessageId(), topic, message.getData().toStringUtf8());
         subscriberMetrics.incrementFailedToConsumeMessage();
+      } finally {
+        consumer.ack();
       }
     };
   }
+
+  private EventMessage deserialise(String json) {
+    EventMessage result = gson.fromJson(json, EventMessage.class);
+    if (result.getEvent() == null && result.getHeader() == null) {
+      Event event = deserialiseEvent(json);
+      result = new EventMessage(new Header(UUID.randomUUID(), event.instanceId), event);
+    }
+    result.validate();
+    return result;
+  }
+
+  private Event deserialiseEvent(String json) {
+    Event event = gson.fromJson(json, Event.class);
+    requireNonNull(event.type, "Event type cannot be null");
+    requireNonNull(event.instanceId, "Event instance id cannot be null");
+    return event;
+  }
 }

src/test/java/com/googlesource/gerrit/plugins/pubsub/PubSubEventSubscriberTest.java

diff --git a/src/test/java/com/googlesource/gerrit/plugins/pubsub/PubSubEventSubscriberTest.java b/src/test/java/com/googlesource/gerrit/plugins/pubsub/PubSubEventSubscriberTest.java
index ec2dd69..c5a8318 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/pubsub/PubSubEventSubscriberTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/pubsub/PubSubEventSubscriberTest.java
@@ -14,17 +14,27 @@
 
 package com.googlesource.gerrit.plugins.pubsub;
 
+import static com.google.common.truth.Truth.assertThat;
+import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.only;
 import static org.mockito.Mockito.verify;
 
 import com.gerritforge.gerrit.eventbroker.EventMessage;
 import com.google.cloud.pubsub.v1.AckReplyConsumer;
 import com.google.cloud.pubsub.v1.MessageReceiver;
-import com.google.gerrit.json.OutputFormat;
+import com.google.gerrit.server.events.Event;
+import com.google.gerrit.server.events.EventGsonProvider;
+import com.google.gerrit.server.events.ProjectCreatedEvent;
+import com.google.gson.Gson;
+import com.google.protobuf.ByteString;
 import com.google.pubsub.v1.PubsubMessage;
+import java.util.UUID;
 import java.util.function.Consumer;
 import org.junit.Test;
 import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Captor;
 import org.mockito.Mock;
 import org.mockito.junit.MockitoJUnitRunner;
 
@@ -35,8 +45,14 @@
   @Mock SubscriberProvider subscriberProviderMock;
   @Mock PubSubSubscriberMetrics pubSubSubscriberMetricsMock;
   @Mock AckReplyConsumer ackReplyConsumerMock;
+  @Mock Consumer<EventMessage> succeedingConsumer;
+  @Captor ArgumentCaptor<EventMessage> eventMessageCaptor;
 
   private static final String TOPIC = "foo";
+  private static final EventMessage eventMessage =
+      new EventMessage(
+          new EventMessage.Header(UUID.randomUUID(), UUID.randomUUID()), new ProjectCreatedEvent());
+  private Gson gson = new EventGsonProvider().get();
 
   @Test
   public void shouldIncrementFailedToConsumeMessageWhenReceivingFails() {
@@ -53,22 +69,69 @@
 
   @Test
   public void shouldIncrementSucceedToConsumeMessageWhenReceivingSucceeds() {
-    Consumer<EventMessage> succeedingConsumer = (message) -> {};
+    PubsubMessage pubsubMessage = sampleMessage();
 
-    messageReceiver(succeedingConsumer)
-        .receiveMessage(PubsubMessage.getDefaultInstance(), ackReplyConsumerMock);
+    messageReceiver(succeedingConsumer).receiveMessage(pubsubMessage, ackReplyConsumerMock);
 
     verify(pubSubSubscriberMetricsMock, only()).incrementSucceedToConsumeMessage();
   }
 
+  @Test
+  public void shouldSkipEventWithoutSourceInstanceId() {
+    Event event = new ProjectCreatedEvent();
+    EventMessage messageWithoutSourceInstanceId =
+        new EventMessage(new EventMessage.Header(UUID.randomUUID(), (String) null), event);
+    PubsubMessage pubsubMessage = sampleMessage(messageWithoutSourceInstanceId);
+
+    messageReceiver(succeedingConsumer).receiveMessage(pubsubMessage, ackReplyConsumerMock);
+
+    verify(succeedingConsumer, never()).accept(messageWithoutSourceInstanceId);
+  }
+
+  @Test
+  public void shouldParseEventObject() {
+    String instanceId = "instance-id";
+
+    Event event = new ProjectCreatedEvent();
+    event.instanceId = instanceId;
+    PubsubMessage pubsubMessage = sampleMessage(event);
+    messageReceiver(succeedingConsumer).receiveMessage(pubsubMessage, ackReplyConsumerMock);
+
+    verify(succeedingConsumer, only()).accept(eventMessageCaptor.capture());
+    EventMessage result = eventMessageCaptor.getValue();
+    assertThat(result.getHeader().sourceInstanceId).isEqualTo(instanceId);
+  }
+
+  @Test
+  public void shouldParseEventObjectWithHeaderAndBodyProjectName() {
+    ProjectCreatedEvent event = new ProjectCreatedEvent();
+    event.instanceId = "instance-id";
+    event.projectName = "header_body_parser_project";
+    PubsubMessage pubsubMessage = sampleMessage(event);
+    messageReceiver(succeedingConsumer).receiveMessage(pubsubMessage, ackReplyConsumerMock);
+
+    verify(succeedingConsumer, only()).accept(any(EventMessage.class));
+  }
+
+  private PubsubMessage sampleMessage(Event event) {
+    String eventPayload = gson.toJson(event);
+    ByteString data = ByteString.copyFromUtf8(eventPayload);
+    return PubsubMessage.newBuilder().setData(data).build();
+  }
+
+  private PubsubMessage sampleMessage(EventMessage message) {
+    String eventPayload = gson.toJson(message);
+    ByteString data = ByteString.copyFromUtf8(eventPayload);
+    return PubsubMessage.newBuilder().setData(data).build();
+  }
+
+  private PubsubMessage sampleMessage() {
+    return sampleMessage(eventMessage);
+  }
+
   private MessageReceiver messageReceiver(Consumer<EventMessage> consumer) {
     return new PubSubEventSubscriber(
-            OutputFormat.JSON_COMPACT.newGson(),
-            subscriberProviderMock,
-            confMock,
-            pubSubSubscriberMetricsMock,
-            TOPIC,
-            consumer)
+            gson, subscriberProviderMock, confMock, pubSubSubscriberMetricsMock, TOPIC, consumer)
         .getMessageReceiver();
   }
 }