Merge branch 'stable-3.2' into stable-3.3 * stable-3.2: Update kafka-client 2.1.0 -> 2.1.1 Change-Id: I827a95bde2afd8439d58d456ee157394ac5d7b5a

diff --git a/BUILD b/BUILD
index c0eab0c..d9f938b 100644
--- a/BUILD
+++ b/BUILD

@@ -17,8 +17,11 @@
     ],
     resources = glob(["src/main/resources/**/*"]),
     deps = [
-        "@kafka-client//jar",
+        "//lib/httpcomponents:httpclient",
         "@events-broker//jar",
+        "@httpasyncclient//jar",
+        "@httpcore-nio//jar",
+        "@kafka-client//jar",
     ],
 )
 
@@ -26,11 +29,12 @@
     name = "events_kafka_tests",
     srcs = glob(["src/test/java/**/*.java"]),
     tags = ["events-kafka"],
+    timeout = "long",
     deps = [
         ":events-kafka__plugin_test_deps",
         "//lib/testcontainers",
-        "@kafka-client//jar",
         "@events-broker//jar",
+        "@kafka-client//jar",
         "@testcontainers-kafka//jar",
     ],
 )

diff --git a/external_plugin_deps.bzl b/external_plugin_deps.bzl
index f2cc67c..794fc10 100644
--- a/external_plugin_deps.bzl
+++ b/external_plugin_deps.bzl

@@ -15,6 +15,6 @@
 
     maven_jar(
         name = "events-broker",
-        artifact = "com.gerritforge:events-broker:3.2.0-rc4",
-        sha1 = "53e3f862ac2c2196dba716756ac9586f4b63af47",
+        artifact = "com.gerritforge:events-broker:3.3.2",
+        sha1 = "d8bcb77047cc12dd7c623b5b4de70a25499d3d6c",
     )

diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/Module.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/Module.java
index 1be52cb..463f1c9 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/kafka/Module.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/Module.java

@@ -18,23 +18,37 @@
 import com.google.gerrit.extensions.events.LifecycleListener;
 import com.google.gerrit.extensions.registration.DynamicSet;
 import com.google.gerrit.server.events.EventListener;
+import com.google.gerrit.server.git.WorkQueue;
 import com.google.gson.Gson;
 import com.google.inject.AbstractModule;
 import com.google.inject.Inject;
+import com.google.inject.Scopes;
 import com.google.inject.Singleton;
 import com.google.inject.TypeLiteral;
+import com.google.inject.assistedinject.FactoryModuleBuilder;
 import com.googlesource.gerrit.plugins.kafka.api.KafkaApiModule;
+import com.googlesource.gerrit.plugins.kafka.config.KafkaProperties;
+import com.googlesource.gerrit.plugins.kafka.config.KafkaProperties.ClientType;
 import com.googlesource.gerrit.plugins.kafka.publish.KafkaPublisher;
+import com.googlesource.gerrit.plugins.kafka.publish.KafkaRestProducer;
+import com.googlesource.gerrit.plugins.kafka.rest.FutureExecutor;
+import com.googlesource.gerrit.plugins.kafka.rest.HttpHostProxy;
+import com.googlesource.gerrit.plugins.kafka.rest.HttpHostProxyProvider;
+import com.googlesource.gerrit.plugins.kafka.rest.KafkaRestClient;
 import com.googlesource.gerrit.plugins.kafka.session.KafkaProducerProvider;
-import org.apache.kafka.clients.producer.KafkaProducer;
+import java.util.concurrent.ExecutorService;
+import org.apache.kafka.clients.producer.Producer;
 
 class Module extends AbstractModule {
-
   private final KafkaApiModule kafkaBrokerModule;
+  private final KafkaProperties kafkaConf;
+  private final WorkQueue workQueue;
 
   @Inject
-  public Module(KafkaApiModule kafkaBrokerModule) {
+  public Module(KafkaApiModule kafkaBrokerModule, KafkaProperties kafkaConf, WorkQueue workQueue) {
     this.kafkaBrokerModule = kafkaBrokerModule;
+    this.kafkaConf = kafkaConf;
+    this.workQueue = workQueue;
   }
 
   @Override
@@ -43,8 +57,25 @@
     DynamicSet.bind(binder(), LifecycleListener.class).to(Manager.class);
     DynamicSet.bind(binder(), EventListener.class).to(KafkaPublisher.class);
 
-    bind(new TypeLiteral<KafkaProducer<String, String>>() {})
-        .toProvider(KafkaProducerProvider.class);
+    ClientType clientType = kafkaConf.getClientType();
+    switch (clientType) {
+      case NATIVE:
+        bind(new TypeLiteral<Producer<String, String>>() {})
+            .toProvider(KafkaProducerProvider.class);
+        break;
+      case REST:
+        bind(ExecutorService.class)
+            .annotatedWith(FutureExecutor.class)
+            .toInstance(
+                workQueue.createQueue(
+                    kafkaConf.getRestApiThreads(), "KafkaRestClientThreadPool", true));
+        bind(HttpHostProxy.class).toProvider(HttpHostProxyProvider.class).in(Scopes.SINGLETON);
+        bind(new TypeLiteral<Producer<String, String>>() {}).to(KafkaRestProducer.class);
+        install(new FactoryModuleBuilder().build(KafkaRestClient.Factory.class));
+        break;
+      default:
+        throw new IllegalArgumentException("Unsupported Kafka client type " + clientType);
+    }
 
     install(kafkaBrokerModule);
   }

diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/api/KafkaApiModule.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/api/KafkaApiModule.java
index 73c7509..47f9969 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/kafka/api/KafkaApiModule.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/api/KafkaApiModule.java

@@ -26,8 +26,12 @@
 import com.google.inject.Singleton;
 import com.google.inject.TypeLiteral;
 import com.googlesource.gerrit.plugins.kafka.broker.ConsumerExecutor;
+import com.googlesource.gerrit.plugins.kafka.config.KafkaProperties.ClientType;
 import com.googlesource.gerrit.plugins.kafka.config.KafkaSubscriberProperties;
 import com.googlesource.gerrit.plugins.kafka.subscribe.KafkaEventDeserializer;
+import com.googlesource.gerrit.plugins.kafka.subscribe.KafkaEventNativeSubscriber;
+import com.googlesource.gerrit.plugins.kafka.subscribe.KafkaEventRestSubscriber;
+import com.googlesource.gerrit.plugins.kafka.subscribe.KafkaEventSubscriber;
 import java.util.Set;
 import java.util.concurrent.ExecutorService;
 import org.apache.kafka.common.serialization.ByteArrayDeserializer;
@@ -54,6 +58,17 @@
 
   @Override
   protected void configure() {
+    ClientType clientType = configuration.getClientType();
+    switch (clientType) {
+      case NATIVE:
+        bind(KafkaEventSubscriber.class).to(KafkaEventNativeSubscriber.class);
+        break;
+      case REST:
+        bind(KafkaEventSubscriber.class).to(KafkaEventRestSubscriber.class);
+        break;
+      default:
+        throw new IllegalArgumentException("Unsupported Kafka client type " + clientType);
+    }
 
     bind(ExecutorService.class)
         .annotatedWith(ConsumerExecutor.class)

diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/config/KafkaProperties.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/config/KafkaProperties.java
index 72d7f91..d8e22a0 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/kafka/config/KafkaProperties.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/config/KafkaProperties.java

@@ -17,43 +17,113 @@
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.CaseFormat;
 import com.google.common.base.Strings;
+import com.google.gerrit.common.Nullable;
 import com.google.gerrit.extensions.annotations.PluginName;
+import com.google.gerrit.server.config.ConfigUtil;
 import com.google.gerrit.server.config.PluginConfig;
 import com.google.gerrit.server.config.PluginConfigFactory;
 import com.google.inject.Inject;
 import com.google.inject.Singleton;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.time.Duration;
 import java.util.UUID;
+import java.util.concurrent.TimeUnit;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.common.serialization.StringSerializer;
 
 @Singleton
 public class KafkaProperties extends java.util.Properties {
+  private static final String PROPERTY_HTTP_WIRE_LOG = "httpWireLog";
+  private static final boolean DEFAULT_HTTP_WIRE_LOG = false;
+  private static final String PROPERTY_REST_API_URI = "restApiUri";
+  private static final String PROPERTY_REST_API_TIMEOUT = "restApiTimeout";
+  private static final Duration DEFAULT_REST_API_TIMEOUT = Duration.ofSeconds(60);
+  private static final String PROPERTY_REST_API_THREADS = "restApiThreads";
+  private static final int DEFAULT_REST_API_THREADS = 10;
+  private static final String PROPERTY_CLIENT_TYPE = "clientType";
+  private static final ClientType DEFAULT_CLIENT_TYPE = ClientType.NATIVE;
+  private static final String PROPERTY_SEND_ASYNC = "sendAsync";
+  private static final boolean DEFAULT_SEND_ASYNC = true;
+  private static final String PROPERTY_STREAM_EVENTS_TOPIC_NAME = "topic";
+  private static final String DEFAULT_STREAM_EVENTS_TOPIC_NAME = "gerrit";
+
   private static final long serialVersionUID = 0L;
 
   public static final String KAFKA_STRING_SERIALIZER = StringSerializer.class.getName();
 
+  public enum ClientType {
+    NATIVE,
+    REST;
+  }
+
   private final String topic;
   private final boolean sendAsync;
+  private final ClientType clientType;
+  private final URI restApiUri;
+  private final boolean httpWireLog;
+  private final Duration restApiTimeout;
+  private final int restApiThreads;
 
   @Inject
   public KafkaProperties(PluginConfigFactory configFactory, @PluginName String pluginName) {
     super();
     setDefaults();
     PluginConfig fromGerritConfig = configFactory.getFromGerritConfig(pluginName);
-    topic = fromGerritConfig.getString("topic", "gerrit");
-    sendAsync = fromGerritConfig.getBoolean("sendAsync", true);
+    topic =
+        fromGerritConfig.getString(
+            PROPERTY_STREAM_EVENTS_TOPIC_NAME, DEFAULT_STREAM_EVENTS_TOPIC_NAME);
+    sendAsync = fromGerritConfig.getBoolean(PROPERTY_SEND_ASYNC, DEFAULT_SEND_ASYNC);
+    clientType = fromGerritConfig.getEnum(PROPERTY_CLIENT_TYPE, DEFAULT_CLIENT_TYPE);
+
+    switch (clientType) {
+      case REST:
+        String restApiUriString = fromGerritConfig.getString(PROPERTY_REST_API_URI);
+        if (Strings.isNullOrEmpty(restApiUriString)) {
+          throw new IllegalArgumentException("Missing REST API URI in Kafka properties");
+        }
+
+        try {
+          restApiUri = new URI(restApiUriString);
+        } catch (URISyntaxException e) {
+          throw new IllegalArgumentException("Invalid Kafka REST API URI: " + restApiUriString, e);
+        }
+        httpWireLog = fromGerritConfig.getBoolean(PROPERTY_HTTP_WIRE_LOG, DEFAULT_HTTP_WIRE_LOG);
+        restApiTimeout =
+            Duration.ofMillis(
+                ConfigUtil.getTimeUnit(
+                    fromGerritConfig.getString(PROPERTY_REST_API_TIMEOUT),
+                    DEFAULT_REST_API_TIMEOUT.toMillis(),
+                    TimeUnit.MILLISECONDS));
+        restApiThreads =
+            fromGerritConfig.getInt(PROPERTY_REST_API_THREADS, DEFAULT_REST_API_THREADS);
+        break;
+      case NATIVE:
+      default:
+        restApiUri = null;
+        httpWireLog = false;
+        restApiTimeout = null;
+        restApiThreads = 0;
+        break;
+    }
+
     applyConfig(fromGerritConfig);
     initDockerizedKafkaServer();
   }
 
   @VisibleForTesting
-  public KafkaProperties(boolean sendAsync) {
+  public KafkaProperties(boolean sendAsync, ClientType clientType, @Nullable URI restApiURI) {
     super();
     setDefaults();
-    topic = "gerrit";
+    topic = DEFAULT_STREAM_EVENTS_TOPIC_NAME;
     this.sendAsync = sendAsync;
+    this.clientType = clientType;
+    this.restApiUri = restApiURI;
     initDockerizedKafkaServer();
+    this.httpWireLog = false;
+    restApiTimeout = DEFAULT_REST_API_TIMEOUT;
+    restApiThreads = DEFAULT_REST_API_THREADS;
   }
 
   private void setDefaults() {
@@ -99,4 +169,24 @@
   public String getBootstrapServers() {
     return getProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG);
   }
+
+  public ClientType getClientType() {
+    return clientType;
+  }
+
+  public URI getRestApiUri() {
+    return restApiUri;
+  }
+
+  public boolean isHttpWireLog() {
+    return httpWireLog;
+  }
+
+  public Duration getRestApiTimeout() {
+    return restApiTimeout;
+  }
+
+  public int getRestApiThreads() {
+    return restApiThreads;
+  }
 }

diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/config/KafkaSubscriberProperties.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/config/KafkaSubscriberProperties.java
index 52d4726..1b0e4db 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/kafka/config/KafkaSubscriberProperties.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/config/KafkaSubscriberProperties.java

@@ -19,6 +19,7 @@
 import com.google.gerrit.server.config.PluginConfigFactory;
 import com.google.inject.Inject;
 import com.google.inject.Singleton;
+import java.net.URI;
 
 @Singleton
 public class KafkaSubscriberProperties extends KafkaProperties {
@@ -43,8 +44,19 @@
   }
 
   @VisibleForTesting
-  public KafkaSubscriberProperties(int pollingInterval, String groupId, int numberOfSubscribers) {
-    super(true);
+  public KafkaSubscriberProperties(
+      int pollingInterval, String groupId, int numberOfSubscribers, ClientType clientType) {
+    this(pollingInterval, groupId, numberOfSubscribers, clientType, null);
+  }
+
+  @VisibleForTesting
+  public KafkaSubscriberProperties(
+      int pollingInterval,
+      String groupId,
+      int numberOfSubscribers,
+      ClientType clientType,
+      URI restApiURI) {
+    super(true, clientType, restApiURI);
     this.pollingInterval = pollingInterval;
     this.groupId = groupId;
     this.numberOfSubscribers = numberOfSubscribers;

diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/publish/KafkaRestProducer.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/publish/KafkaRestProducer.java
new file mode 100644
index 0000000..8799cef
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/publish/KafkaRestProducer.java

@@ -0,0 +1,137 @@
+// Copyright (C) 2021 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.googlesource.gerrit.plugins.kafka.publish;
+
+import com.google.common.flogger.FluentLogger;
+import com.google.common.util.concurrent.Futures;
+import com.google.inject.Inject;
+import com.googlesource.gerrit.plugins.kafka.config.KafkaProperties;
+import com.googlesource.gerrit.plugins.kafka.rest.KafkaRestClient;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.apache.http.HttpStatus;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.entity.ContentType;
+import org.apache.http.entity.StringEntity;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.clients.producer.Callback;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.Metric;
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.ProducerFencedException;
+
+public class KafkaRestProducer implements Producer<String, String> {
+  private static final RecordMetadata ZEROS_RECORD_METADATA =
+      new RecordMetadata(null, 0, 0, 0, null, 0, 0);
+  private static final FluentLogger logger = FluentLogger.forEnclosingClass();
+  private static final String KAFKA_V2_JSON = "application/vnd.kafka.json.v2+json";
+  private final KafkaRestClient restClient;
+
+  @Inject
+  public KafkaRestProducer(KafkaProperties kafkaConf, KafkaRestClient.Factory restClientFactory) {
+    restClient = restClientFactory.create(kafkaConf);
+  }
+
+  @Override
+  public void initTransactions() {
+    unsupported();
+  }
+
+  @Override
+  public void beginTransaction() throws ProducerFencedException {
+    unsupported();
+  }
+
+  @Override
+  public void sendOffsetsToTransaction(
+      Map<TopicPartition, OffsetAndMetadata> offsets, String consumerGroupId)
+      throws ProducerFencedException {
+    unsupported();
+  }
+
+  @Override
+  public void commitTransaction() throws ProducerFencedException {
+    unsupported();
+  }
+
+  @Override
+  public void abortTransaction() throws ProducerFencedException {
+    unsupported();
+  }
+
+  @Override
+  public Future<RecordMetadata> send(ProducerRecord<String, String> record) {
+    return send(record, null);
+  }
+
+  @Override
+  public Future<RecordMetadata> send(ProducerRecord<String, String> record, Callback callback) {
+    HttpPost post =
+        restClient.createPostToTopic(
+            record.topic(),
+            new StringEntity(
+                getRecordAsJson(record),
+                ContentType.create(KAFKA_V2_JSON, StandardCharsets.UTF_8)));
+    return restClient.mapAsync(
+        restClient.execute(post, HttpStatus.SC_OK),
+        (res) -> Futures.immediateFuture(ZEROS_RECORD_METADATA));
+  }
+
+  @Override
+  public void flush() {
+    unsupported();
+  }
+
+  @Override
+  public List<PartitionInfo> partitionsFor(String topic) {
+    return unsupported();
+  }
+
+  @Override
+  public Map<MetricName, ? extends Metric> metrics() {
+    return unsupported();
+  }
+
+  @Override
+  public void close() {
+    try {
+      restClient.close();
+    } catch (IOException e) {
+      logger.atWarning().withCause(e).log("Unable to close httpclient");
+    }
+  }
+
+  @Override
+  public void close(long timeout, TimeUnit unit) {
+    close();
+  }
+
+  private String getRecordAsJson(ProducerRecord<String, String> record) {
+    return String.format(
+        "{\"records\":[{\"key\":\"%s\",\"value\":%s}]}", record.key(), record.value());
+  }
+
+  private <T> T unsupported() {
+    throw new IllegalArgumentException("Unsupported method");
+  }
+}

diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/rest/FutureExecutor.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/rest/FutureExecutor.java
new file mode 100644
index 0000000..2ff0840
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/rest/FutureExecutor.java

@@ -0,0 +1,23 @@
+// Copyright (C) 2021 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.googlesource.gerrit.plugins.kafka.rest;
+
+import com.google.inject.BindingAnnotation;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+
+@Retention(RetentionPolicy.RUNTIME)
+@BindingAnnotation
+public @interface FutureExecutor {}

diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/rest/HttpHostProxy.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/rest/HttpHostProxy.java
new file mode 100644
index 0000000..595b95a
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/rest/HttpHostProxy.java

@@ -0,0 +1,56 @@
+// Copyright (C) 2021 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.googlesource.gerrit.plugins.kafka.rest;
+
+import com.google.gerrit.common.Nullable;
+import java.net.URL;
+import org.apache.http.HttpHost;
+import org.apache.http.auth.AuthScope;
+import org.apache.http.auth.UsernamePasswordCredentials;
+import org.apache.http.client.CredentialsProvider;
+import org.apache.http.client.config.RequestConfig.Builder;
+import org.apache.http.impl.client.BasicCredentialsProvider;
+import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
+
+public class HttpHostProxy {
+  private final URL proxyUrl;
+  private final String username;
+  private final String password;
+
+  public HttpHostProxy(URL proxyUrl, @Nullable String username, @Nullable String password) {
+    this.proxyUrl = proxyUrl;
+    this.username = username;
+    this.password = password;
+  }
+
+  public Builder apply(Builder clientBuilder) {
+    if (proxyUrl != null) {
+      clientBuilder.setProxy(
+          new HttpHost(proxyUrl.getHost(), proxyUrl.getPort(), proxyUrl.getProtocol()));
+    }
+    return clientBuilder;
+  }
+
+  public HttpAsyncClientBuilder apply(HttpAsyncClientBuilder custom) {
+    if (proxyUrl != null && username != null && password != null) {
+      CredentialsProvider credsProvider = new BasicCredentialsProvider();
+      credsProvider.setCredentials(
+          new AuthScope(proxyUrl.getHost(), proxyUrl.getPort()),
+          new UsernamePasswordCredentials(username, password));
+      custom.setDefaultCredentialsProvider(credsProvider);
+    }
+    return custom;
+  }
+}

diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/rest/HttpHostProxyProvider.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/rest/HttpHostProxyProvider.java
new file mode 100644
index 0000000..7c82ecd
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/rest/HttpHostProxyProvider.java

@@ -0,0 +1,54 @@
+// Copyright (C) 2021 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.googlesource.gerrit.plugins.kafka.rest;
+
+import com.google.common.base.Strings;
+import com.google.gerrit.server.config.GerritServerConfig;
+import com.google.inject.Inject;
+import com.google.inject.Provider;
+import java.net.MalformedURLException;
+import java.net.URL;
+import org.eclipse.jgit.lib.Config;
+
+public class HttpHostProxyProvider implements Provider<HttpHostProxy> {
+  private URL proxyUrl;
+  private String proxyUser;
+  private String proxyPassword;
+
+  @Inject
+  HttpHostProxyProvider(@GerritServerConfig Config config) throws MalformedURLException {
+    String proxyUrlStr = config.getString("http", null, "proxy");
+    if (!Strings.isNullOrEmpty(proxyUrlStr)) {
+      proxyUrl = new URL(proxyUrlStr);
+      proxyUser = config.getString("http", null, "proxyUsername");
+      proxyPassword = config.getString("http", null, "proxyPassword");
+      String userInfo = proxyUrl.getUserInfo();
+      if (userInfo != null) {
+        int c = userInfo.indexOf(':');
+        if (0 < c) {
+          proxyUser = userInfo.substring(0, c);
+          proxyPassword = userInfo.substring(c + 1);
+        } else {
+          proxyUser = userInfo;
+        }
+      }
+    }
+  }
+
+  @Override
+  public HttpHostProxy get() {
+    return new HttpHostProxy(proxyUrl, proxyUser, proxyPassword);
+  }
+}

diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/rest/KafkaRestClient.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/rest/KafkaRestClient.java
new file mode 100644
index 0000000..b22ae7a
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/rest/KafkaRestClient.java

@@ -0,0 +1,254 @@
+// Copyright (C) 2021 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.googlesource.gerrit.plugins.kafka.rest;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+
+import com.google.common.base.Function;
+import com.google.common.flogger.FluentLogger;
+import com.google.common.net.MediaType;
+import com.google.common.util.concurrent.AsyncFunction;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.JdkFutureAdapters;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.gerrit.common.Nullable;
+import com.google.inject.Inject;
+import com.google.inject.assistedinject.Assisted;
+import com.googlesource.gerrit.plugins.kafka.config.KafkaProperties;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.net.URI;
+import java.net.URLEncoder;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.stream.Collectors;
+import org.apache.http.HttpEntity;
+import org.apache.http.HttpHeaders;
+import org.apache.http.HttpResponse;
+import org.apache.http.HttpStatus;
+import org.apache.http.client.config.RequestConfig;
+import org.apache.http.client.config.RequestConfig.Builder;
+import org.apache.http.client.methods.HttpDelete;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.client.methods.HttpRequestBase;
+import org.apache.http.entity.ContentType;
+import org.apache.http.entity.StringEntity;
+import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
+import org.apache.http.impl.nio.client.HttpAsyncClients;
+import org.apache.log4j.Level;
+import org.apache.log4j.Logger;
+
+public class KafkaRestClient {
+  private static final FluentLogger logger = FluentLogger.forEnclosingClass();
+  private static final String KAFKA_V2_JSON = "application/vnd.kafka.json.v2+json";
+  private static final String KAFKA_V2 = "application/vnd.kafka.v2+json";
+
+  private final HttpHostProxy proxy;
+  private final CloseableHttpAsyncClient httpclient;
+  private final URI kafkaRestApiUri;
+  private final ExecutorService futureExecutor;
+  private final int kafkaRestApiTimeoutMsec;
+
+  public interface Factory {
+    KafkaRestClient create(KafkaProperties configuration);
+  }
+
+  @Inject
+  public KafkaRestClient(
+      HttpHostProxy httpHostProxy,
+      @FutureExecutor ExecutorService executor,
+      @Assisted KafkaProperties configuration) {
+    proxy = httpHostProxy;
+    httpclient = proxy.apply(HttpAsyncClients.custom()).build();
+    httpclient.start();
+    kafkaRestApiUri = configuration.getRestApiUri();
+    kafkaRestApiTimeoutMsec = (int) configuration.getRestApiTimeout().toMillis();
+    if (configuration.isHttpWireLog()) {
+      enableHttpWireLog();
+    }
+    this.futureExecutor = executor;
+  }
+
+  public static void enableHttpWireLog() {
+    Logger.getLogger("org.apache.http.wire").setLevel(Level.TRACE);
+  }
+
+  public ListenableFuture<HttpResponse> execute(HttpRequestBase request, int... expectedStatuses) {
+    return Futures.transformAsync(
+        listenableFutureOf(httpclient.execute(request, null)),
+        (res) -> {
+          IOException exc =
+              getResponseException(
+                  String.format("HTTP %s %s FAILED", request.getMethod(), request.getURI()),
+                  res,
+                  expectedStatuses);
+          if (exc == null) {
+            return Futures.immediateFuture(res);
+          }
+          return Futures.immediateFailedFuture(exc);
+        },
+        futureExecutor);
+  }
+
+  public <I, O> ListenableFuture<O> mapAsync(
+      ListenableFuture<I> inputFuture, AsyncFunction<? super I, ? extends O> mapFunction) {
+    return Futures.transformAsync(inputFuture, mapFunction, futureExecutor);
+  }
+
+  public <I, O> ListenableFuture<O> map(
+      ListenableFuture<I> inputFuture, Function<? super I, ? extends O> mapFunction) {
+    return Futures.transform(inputFuture, mapFunction, futureExecutor);
+  }
+
+  public HttpGet createGetTopic(String topic) {
+    HttpGet get = new HttpGet(kafkaRestApiUri.resolve("/topics/" + topic));
+    get.addHeader(HttpHeaders.ACCEPT, KAFKA_V2_JSON);
+    get.setConfig(createRequestConfig());
+    return get;
+  }
+
+  public HttpGet createGetRecords(URI consumerUri) {
+    HttpGet get = new HttpGet(consumerUri.resolve(consumerUri.getPath() + "/records"));
+    get.addHeader(HttpHeaders.ACCEPT, KAFKA_V2_JSON);
+    get.setConfig(createRequestConfig());
+    return get;
+  }
+
+  public HttpPost createPostToConsumer(String consumerGroup) {
+    HttpPost post =
+        new HttpPost(
+            kafkaRestApiUri.resolve(
+                kafkaRestApiUri.getPath()
+                    + "/consumers/"
+                    + URLEncoder.encode(consumerGroup, UTF_8)));
+    post.addHeader(HttpHeaders.ACCEPT, MediaType.ANY_TYPE.toString());
+    post.setConfig(createRequestConfig());
+    post.setEntity(
+        new StringEntity(
+            "{\"format\": \"json\",\"auto.offset.reset\": \"earliest\"}",
+            ContentType.create(KAFKA_V2, UTF_8)));
+    return post;
+  }
+
+  public HttpDelete createDeleteToConsumer(URI consumerUri) {
+    HttpDelete delete = new HttpDelete(consumerUri);
+    delete.addHeader(HttpHeaders.ACCEPT, "*/*");
+    delete.setConfig(createRequestConfig());
+    return delete;
+  }
+
+  public HttpPost createPostToSubscribe(URI consumerUri, String topic) {
+    HttpPost post = new HttpPost(consumerUri.resolve(consumerUri.getPath() + "/subscription"));
+    post.addHeader(HttpHeaders.ACCEPT, "*/*");
+    post.setConfig(createRequestConfig());
+    post.setEntity(
+        new StringEntity(
+            String.format("{\"topics\":[\"%s\"]}", topic), ContentType.create(KAFKA_V2, UTF_8)));
+    return post;
+  }
+
+  public HttpPost createPostToTopic(String topic, HttpEntity postBodyEntity) {
+    HttpPost post =
+        new HttpPost(kafkaRestApiUri.resolve("/topics/" + URLEncoder.encode(topic, UTF_8)));
+    post.addHeader(HttpHeaders.ACCEPT, "*/*");
+    post.setConfig(createRequestConfig());
+    post.setEntity(postBodyEntity);
+    return post;
+  }
+
+  public HttpPost createPostSeekTopicFromBeginning(
+      URI consumerInstanceURI, String topic, Set<Integer> partitions) {
+    HttpPost post = new HttpPost(consumerInstanceURI.resolve("/positions/beginning"));
+    post.addHeader(HttpHeaders.ACCEPT, "*/*");
+    post.setConfig(createRequestConfig());
+    post.setEntity(
+        new StringEntity(
+            String.format(
+                "{\"partitions\",[%s]}",
+                partitions.stream()
+                    .map(
+                        partition ->
+                            String.format("{\"topic\":\"%s\",\"partition\":%d}", topic, partition))
+                    .collect(Collectors.joining(","))),
+            UTF_8));
+    return post;
+  }
+
+  @Nullable
+  public IOException getResponseException(
+      String errorMessage, HttpResponse response, int... okHttpStatuses) {
+    int responseHttpStatus = response.getStatusLine().getStatusCode();
+    if (okHttpStatuses.length == 0) {
+      okHttpStatuses =
+          new int[] {HttpStatus.SC_OK, HttpStatus.SC_CREATED, HttpStatus.SC_NO_CONTENT};
+    }
+    for (int httpStatus : okHttpStatuses) {
+      if (responseHttpStatus == httpStatus) {
+        return null;
+      }
+    }
+
+    String responseBody = "";
+    try {
+      responseBody = getStringEntity(response);
+    } catch (IOException e) {
+      logger.atWarning().withCause(e).log(
+          "Unable to extrace the string entity for response %d (%s)",
+          response.getStatusLine().getStatusCode(), response.getStatusLine().getReasonPhrase());
+    }
+
+    return new IOException(
+        String.format(
+            "%s\nHTTP status %d (%s)\n%s",
+            errorMessage,
+            response.getStatusLine().getStatusCode(),
+            response.getStatusLine().getReasonPhrase(),
+            responseBody));
+  }
+
+  protected String getStringEntity(HttpResponse response) throws IOException {
+    HttpEntity entity = response.getEntity();
+    try (ByteArrayOutputStream outStream = new ByteArrayOutputStream()) {
+      entity.writeTo(outStream);
+      outStream.close();
+      return outStream.toString(UTF_8);
+    }
+  }
+
+  private <V> ListenableFuture<V> listenableFutureOf(Future<V> future) {
+    return JdkFutureAdapters.listenInPoolThread(future, futureExecutor);
+  }
+
+  private RequestConfig createRequestConfig() {
+    Builder configBuilder =
+        RequestConfig.custom()
+            .setConnectionRequestTimeout(kafkaRestApiTimeoutMsec)
+            .setConnectTimeout(kafkaRestApiTimeoutMsec)
+            .setSocketTimeout(kafkaRestApiTimeoutMsec);
+    configBuilder = proxy.apply(configBuilder);
+    RequestConfig config = configBuilder.build();
+    return config;
+  }
+
+  public void close() throws IOException {
+    httpclient.close();
+  }
+
+  public URI resolveURI(String path) {
+    return kafkaRestApiUri.resolve(path);
+  }
+}

diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/session/KafkaProducerProvider.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/session/KafkaProducerProvider.java
index b1f11f7..4fb98b2 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/kafka/session/KafkaProducerProvider.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/session/KafkaProducerProvider.java

@@ -18,8 +18,9 @@
 import com.google.inject.Provider;
 import com.googlesource.gerrit.plugins.kafka.config.KafkaProperties;
 import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.Producer;
 
-public class KafkaProducerProvider implements Provider<KafkaProducer<String, String>> {
+public class KafkaProducerProvider implements Provider<Producer<String, String>> {
   private final KafkaProperties properties;
 
   @Inject
@@ -28,7 +29,7 @@
   }
 
   @Override
-  public KafkaProducer<String, String> get() {
+  public Producer<String, String> get() {
     return new KafkaProducer<>(properties);
   }
 }

diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/session/KafkaSession.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/session/KafkaSession.java
index bb79cb5..fbaef6f 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/kafka/session/KafkaSession.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/session/KafkaSession.java

@@ -18,8 +18,8 @@
 import com.google.inject.Provider;
 import com.googlesource.gerrit.plugins.kafka.config.KafkaProperties;
 import com.googlesource.gerrit.plugins.kafka.publish.KafkaEventsPublisherMetrics;
+import java.net.URI;
 import java.util.concurrent.Future;
-import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.clients.producer.RecordMetadata;
@@ -30,13 +30,13 @@
 
   private static final Logger LOGGER = LoggerFactory.getLogger(KafkaSession.class);
   private final KafkaProperties properties;
-  private final Provider<KafkaProducer<String, String>> producerProvider;
+  private final Provider<Producer<String, String>> producerProvider;
   private final KafkaEventsPublisherMetrics publisherMetrics;
   private volatile Producer<String, String> producer;
 
   @Inject
   public KafkaSession(
-      Provider<KafkaProducer<String, String>> producerProvider,
+      Provider<Producer<String, String>> producerProvider,
       KafkaProperties properties,
       KafkaEventsPublisherMetrics publisherMetrics) {
     this.producerProvider = producerProvider;
@@ -57,12 +57,37 @@
       return;
     }
 
-    LOGGER.info("Connect to {}...", properties.getProperty("bootstrap.servers"));
-    /* Need to make sure that the thread of the running connection uses
-     * the correct class loader otherwize you can endup with hard to debug
-     * ClassNotFoundExceptions
-     */
-    setConnectionClassLoader();
+    switch (properties.getClientType()) {
+      case NATIVE:
+        String bootstrapServers = properties.getProperty("bootstrap.servers");
+        if (bootstrapServers == null) {
+          LOGGER.warn("No Kafka bootstrap.servers property defined: session not started.");
+          return;
+        }
+
+        LOGGER.info("Connect to {}...", bootstrapServers);
+        /* Need to make sure that the thread of the running connection uses
+         * the correct class loader otherwise you can end up with hard to debug
+         * ClassNotFoundExceptions
+         */
+        setConnectionClassLoader();
+        break;
+
+      case REST:
+        URI kafkaProxyUri = properties.getRestApiUri();
+        if (kafkaProxyUri == null) {
+          LOGGER.warn("No Kafka Proxy URL property defined: session not started.");
+          return;
+        }
+
+        LOGGER.info("Connect to {}...", kafkaProxyUri);
+        break;
+
+      default:
+        LOGGER.error("Unsupported Kafka Client Type %s", properties.getClientType());
+        return;
+    }
+
     producer = producerProvider.get();
     LOGGER.info("Connection established.");
   }

diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventDeserializer.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventDeserializer.java
index bab2ad0..4c57a54 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventDeserializer.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventDeserializer.java

@@ -14,11 +14,16 @@
 
 package com.googlesource.gerrit.plugins.kafka.subscribe;
 
+import static java.util.Objects.requireNonNull;
+
 import com.gerritforge.gerrit.eventbroker.EventMessage;
+import com.gerritforge.gerrit.eventbroker.EventMessage.Header;
+import com.google.gerrit.server.events.Event;
 import com.google.gson.Gson;
 import com.google.inject.Inject;
 import com.google.inject.Singleton;
 import java.util.Map;
+import java.util.UUID;
 import org.apache.kafka.common.serialization.Deserializer;
 import org.apache.kafka.common.serialization.StringDeserializer;
 
@@ -42,13 +47,23 @@
 
   @Override
   public EventMessage deserialize(String topic, byte[] data) {
-    final EventMessage result =
-        gson.fromJson(stringDeserializer.deserialize(topic, data), EventMessage.class);
+    String json = stringDeserializer.deserialize(topic, data);
+    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;
+  }
+
   @Override
   public void close() {}
 }

diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventNativeSubscriber.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventNativeSubscriber.java
new file mode 100644
index 0000000..a98e098
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventNativeSubscriber.java

@@ -0,0 +1,207 @@
+// Copyright (C) 2019 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package com.googlesource.gerrit.plugins.kafka.subscribe;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+
+import com.gerritforge.gerrit.eventbroker.EventMessage;
+import com.google.common.flogger.FluentLogger;
+import com.google.gerrit.server.util.ManualRequestContext;
+import com.google.gerrit.server.util.OneOffRequestContext;
+import com.google.inject.Inject;
+import com.googlesource.gerrit.plugins.kafka.broker.ConsumerExecutor;
+import com.googlesource.gerrit.plugins.kafka.config.KafkaSubscriberProperties;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.Random;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.common.errors.WakeupException;
+import org.apache.kafka.common.serialization.Deserializer;
+
+public class KafkaEventNativeSubscriber implements KafkaEventSubscriber {
+  private static final FluentLogger logger = FluentLogger.forEnclosingClass();
+  private static final int DELAY_RECONNECT_AFTER_FAILURE_MSEC = 1000;
+
+  private final OneOffRequestContext oneOffCtx;
+  private final AtomicBoolean closed = new AtomicBoolean(false);
+
+  private final Deserializer<EventMessage> valueDeserializer;
+  private final KafkaSubscriberProperties configuration;
+  private final ExecutorService executor;
+  private final KafkaEventSubscriberMetrics subscriberMetrics;
+  private final KafkaConsumerFactory consumerFactory;
+  private final Deserializer<byte[]> keyDeserializer;
+
+  private java.util.function.Consumer<EventMessage> messageProcessor;
+  private String topic;
+  private AtomicBoolean resetOffset = new AtomicBoolean(false);
+
+  private volatile ReceiverJob receiver;
+
+  @Inject
+  public KafkaEventNativeSubscriber(
+      KafkaSubscriberProperties configuration,
+      KafkaConsumerFactory consumerFactory,
+      Deserializer<byte[]> keyDeserializer,
+      Deserializer<EventMessage> valueDeserializer,
+      OneOffRequestContext oneOffCtx,
+      @ConsumerExecutor ExecutorService executor,
+      KafkaEventSubscriberMetrics subscriberMetrics) {
+
+    this.configuration = configuration;
+    this.oneOffCtx = oneOffCtx;
+    this.executor = executor;
+    this.subscriberMetrics = subscriberMetrics;
+    this.consumerFactory = consumerFactory;
+    this.keyDeserializer = keyDeserializer;
+    this.valueDeserializer = valueDeserializer;
+  }
+
+  /* (non-Javadoc)
+   * @see com.googlesource.gerrit.plugins.kafka.subscribe.KafkaEventSubscriber#subscribe(java.lang.String, java.util.function.Consumer)
+   */
+  @Override
+  public void subscribe(String topic, java.util.function.Consumer<EventMessage> messageProcessor) {
+    this.topic = topic;
+    this.messageProcessor = messageProcessor;
+    logger.atInfo().log(
+        "Kafka consumer subscribing to topic alias [%s] for event topic [%s]", topic, topic);
+    runReceiver();
+  }
+
+  private void runReceiver() {
+    final ClassLoader previousClassLoader = Thread.currentThread().getContextClassLoader();
+    try {
+      Thread.currentThread()
+          .setContextClassLoader(KafkaEventNativeSubscriber.class.getClassLoader());
+      Consumer<byte[], byte[]> consumer = consumerFactory.create(keyDeserializer);
+      consumer.subscribe(Collections.singleton(topic));
+      receiver = new ReceiverJob(consumer);
+      executor.execute(receiver);
+    } finally {
+      Thread.currentThread().setContextClassLoader(previousClassLoader);
+    }
+  }
+
+  /* (non-Javadoc)
+   * @see com.googlesource.gerrit.plugins.kafka.subscribe.KafkaEventSubscriber#shutdown()
+   */
+  @Override
+  public void shutdown() {
+    closed.set(true);
+    receiver.wakeup();
+  }
+
+  /* (non-Javadoc)
+   * @see com.googlesource.gerrit.plugins.kafka.subscribe.KafkaEventSubscriber#getMessageProcessor()
+   */
+  @Override
+  public java.util.function.Consumer<EventMessage> getMessageProcessor() {
+    return messageProcessor;
+  }
+
+  /* (non-Javadoc)
+   * @see com.googlesource.gerrit.plugins.kafka.subscribe.KafkaEventSubscriber#getTopic()
+   */
+  @Override
+  public String getTopic() {
+    return topic;
+  }
+
+  /* (non-Javadoc)
+   * @see com.googlesource.gerrit.plugins.kafka.subscribe.KafkaEventSubscriber#resetOffset()
+   */
+  @Override
+  public void resetOffset() {
+    resetOffset.set(true);
+  }
+
+  private class ReceiverJob implements Runnable {
+    private final Consumer<byte[], byte[]> consumer;
+
+    public ReceiverJob(Consumer<byte[], byte[]> consumer) {
+      this.consumer = consumer;
+    }
+
+    public void wakeup() {
+      consumer.wakeup();
+    }
+
+    @Override
+    public void run() {
+      try {
+        consume();
+      } catch (Exception e) {
+        logger.atSevere().withCause(e).log("Consumer loop of topic %s ended", topic);
+      }
+    }
+
+    private void consume() throws InterruptedException {
+      try {
+        while (!closed.get()) {
+          if (resetOffset.getAndSet(false)) {
+            // Make sure there is an assignment for this consumer
+            while (consumer.assignment().isEmpty() && !closed.get()) {
+              logger.atInfo().log(
+                  "Resetting offset: no partitions assigned to the consumer, request assignment.");
+              consumer.poll(Duration.ofMillis(configuration.getPollingInterval()));
+            }
+            consumer.seekToBeginning(consumer.assignment());
+          }
+          ConsumerRecords<byte[], byte[]> consumerRecords =
+              consumer.poll(Duration.ofMillis(configuration.getPollingInterval()));
+          consumerRecords.forEach(
+              consumerRecord -> {
+                try (ManualRequestContext ctx = oneOffCtx.open()) {
+                  EventMessage event =
+                      valueDeserializer.deserialize(consumerRecord.topic(), consumerRecord.value());
+                  messageProcessor.accept(event);
+                } catch (Exception e) {
+                  logger.atSevere().withCause(e).log(
+                      "Malformed event '%s': [Exception: %s]",
+                      new String(consumerRecord.value(), UTF_8));
+                  subscriberMetrics.incrementSubscriberFailedToConsumeMessage();
+                }
+              });
+        }
+      } catch (WakeupException e) {
+        // Ignore exception if closing
+        if (!closed.get()) {
+          logger.atSevere().withCause(e).log("Consumer loop of topic %s interrupted", topic);
+          reconnectAfterFailure();
+        }
+      } catch (Exception e) {
+        subscriberMetrics.incrementSubscriberFailedToPollMessages();
+        logger.atSevere().withCause(e).log(
+            "Existing consumer loop of topic %s because of a non-recoverable exception", topic);
+        reconnectAfterFailure();
+      } finally {
+        consumer.close();
+      }
+    }
+
+    private void reconnectAfterFailure() throws InterruptedException {
+      // Random delay with average of DELAY_RECONNECT_AFTER_FAILURE_MSEC
+      // for avoiding hammering exactly at the same interval in case of failure
+      long reconnectDelay =
+          DELAY_RECONNECT_AFTER_FAILURE_MSEC / 2
+              + new Random().nextInt(DELAY_RECONNECT_AFTER_FAILURE_MSEC);
+      Thread.sleep(reconnectDelay);
+      runReceiver();
+    }
+  }
+}

diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventRestSubscriber.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventRestSubscriber.java
new file mode 100644
index 0000000..c792ca8
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventRestSubscriber.java

@@ -0,0 +1,339 @@
+// Copyright (C) 2021 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package com.googlesource.gerrit.plugins.kafka.subscribe;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+
+import com.gerritforge.gerrit.eventbroker.EventMessage;
+import com.google.common.flogger.FluentLogger;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.gerrit.server.util.ManualRequestContext;
+import com.google.gerrit.server.util.OneOffRequestContext;
+import com.google.gson.Gson;
+import com.google.gson.JsonArray;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+import com.google.inject.Inject;
+import com.googlesource.gerrit.plugins.kafka.broker.ConsumerExecutor;
+import com.googlesource.gerrit.plugins.kafka.config.KafkaSubscriberProperties;
+import com.googlesource.gerrit.plugins.kafka.rest.KafkaRestClient;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.Reader;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import java.util.stream.StreamSupport;
+import org.apache.http.HttpResponse;
+import org.apache.http.HttpStatus;
+import org.apache.http.client.methods.HttpDelete;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.serialization.Deserializer;
+
+public class KafkaEventRestSubscriber implements KafkaEventSubscriber {
+  private static final FluentLogger logger = FluentLogger.forEnclosingClass();
+  private static final int DELAY_RECONNECT_AFTER_FAILURE_MSEC = 1000;
+
+  private final OneOffRequestContext oneOffCtx;
+  private final AtomicBoolean closed = new AtomicBoolean(false);
+
+  private final Deserializer<EventMessage> valueDeserializer;
+  private final KafkaSubscriberProperties configuration;
+  private final ExecutorService executor;
+  private final KafkaEventSubscriberMetrics subscriberMetrics;
+  private final Gson gson;
+
+  private java.util.function.Consumer<EventMessage> messageProcessor;
+  private String topic;
+  private final KafkaRestClient restClient;
+  private final AtomicBoolean resetOffset;
+  private final long restClientTimeoutMs;
+  private volatile ReceiverJob receiver;
+
+  @Inject
+  public KafkaEventRestSubscriber(
+      KafkaSubscriberProperties configuration,
+      Deserializer<EventMessage> valueDeserializer,
+      OneOffRequestContext oneOffCtx,
+      @ConsumerExecutor ExecutorService executor,
+      KafkaEventSubscriberMetrics subscriberMetrics,
+      KafkaRestClient.Factory restClientFactory) {
+
+    this.configuration = configuration;
+    this.oneOffCtx = oneOffCtx;
+    this.executor = executor;
+    this.subscriberMetrics = subscriberMetrics;
+    this.valueDeserializer = valueDeserializer;
+
+    gson = new Gson();
+    restClient = restClientFactory.create(configuration);
+    resetOffset = new AtomicBoolean(false);
+    restClientTimeoutMs = configuration.getRestApiTimeout().toMillis();
+  }
+
+  /* (non-Javadoc)
+   * @see com.googlesource.gerrit.plugins.kafka.subscribe.KafkaEventSubscriber#subscribe(java.lang.String, java.util.function.Consumer)
+   */
+  @Override
+  public void subscribe(String topic, java.util.function.Consumer<EventMessage> messageProcessor) {
+    this.topic = topic;
+    this.messageProcessor = messageProcessor;
+    logger.atInfo().log(
+        "Kafka consumer subscribing to topic alias [%s] for event topic [%s]", topic, topic);
+    try {
+      runReceiver();
+    } catch (InterruptedException | ExecutionException | TimeoutException e) {
+      throw new IllegalStateException(e);
+    }
+  }
+
+  private void runReceiver() throws InterruptedException, ExecutionException, TimeoutException {
+    receiver = new ReceiverJob(configuration.getGroupId());
+    executor.execute(receiver);
+  }
+
+  /* (non-Javadoc)
+   * @see com.googlesource.gerrit.plugins.kafka.subscribe.KafkaEventSubscriber#shutdown()
+   */
+  @Override
+  public void shutdown() {
+    try {
+      closed.set(true);
+      receiver.close();
+    } catch (InterruptedException | ExecutionException | IOException | TimeoutException e) {
+      logger.atWarning().withCause(e).log("Unable to close receiver for topic=%s", topic);
+    }
+  }
+
+  /* (non-Javadoc)
+   * @see com.googlesource.gerrit.plugins.kafka.subscribe.KafkaEventSubscriber#getMessageProcessor()
+   */
+  @Override
+  public java.util.function.Consumer<EventMessage> getMessageProcessor() {
+    return messageProcessor;
+  }
+
+  /* (non-Javadoc)
+   * @see com.googlesource.gerrit.plugins.kafka.subscribe.KafkaEventSubscriber#getTopic()
+   */
+  @Override
+  public String getTopic() {
+    return topic;
+  }
+
+  /* (non-Javadoc)
+   * @see com.googlesource.gerrit.plugins.kafka.subscribe.KafkaEventSubscriber#resetOffset()
+   */
+  @Override
+  public void resetOffset() {
+    resetOffset.set(true);
+  }
+
+  private class ReceiverJob implements Runnable {
+    private final ListenableFuture<URI> kafkaRestConsumerUri;
+    private final ListenableFuture<?> kafkaSubscriber;
+
+    public ReceiverJob(String consumerGroup)
+        throws InterruptedException, ExecutionException, TimeoutException {
+      kafkaRestConsumerUri = createConsumer(consumerGroup);
+      kafkaSubscriber = restClient.mapAsync(kafkaRestConsumerUri, this::subscribeToTopic);
+      kafkaSubscriber.get(restClientTimeoutMs, TimeUnit.MILLISECONDS);
+    }
+
+    public void close()
+        throws InterruptedException, ExecutionException, IOException, TimeoutException {
+      restClient
+          .mapAsync(kafkaRestConsumerUri, this::deleteConsumer)
+          .get(restClientTimeoutMs, TimeUnit.MILLISECONDS);
+      restClient.close();
+    }
+
+    @Override
+    public void run() {
+      try {
+        consume();
+      } catch (Exception e) {
+        logger.atSevere().withCause(e).log("Consumer loop of topic %s ended", topic);
+      }
+    }
+
+    private void consume() throws InterruptedException, ExecutionException, TimeoutException {
+      try {
+        while (!closed.get()) {
+          if (resetOffset.getAndSet(false)) {
+            restClient
+                .map(getTopicPartitions(), this::seekToBeginning)
+                .get(restClientTimeoutMs, TimeUnit.MILLISECONDS);
+          }
+
+          ConsumerRecords<byte[], byte[]> records =
+              restClient
+                  .mapAsync(kafkaRestConsumerUri, this::getRecords)
+                  .get(restClientTimeoutMs, TimeUnit.MILLISECONDS);
+          records.forEach(
+              consumerRecord -> {
+                try (ManualRequestContext ctx = oneOffCtx.open()) {
+                  EventMessage event =
+                      valueDeserializer.deserialize(consumerRecord.topic(), consumerRecord.value());
+                  messageProcessor.accept(event);
+                } catch (Exception e) {
+                  logger.atSevere().withCause(e).log(
+                      "Malformed event '%s': [Exception: %s]",
+                      new String(consumerRecord.value(), UTF_8));
+                  subscriberMetrics.incrementSubscriberFailedToConsumeMessage();
+                }
+              });
+        }
+      } catch (Exception e) {
+        subscriberMetrics.incrementSubscriberFailedToPollMessages();
+        logger.atSevere().withCause(e).log(
+            "Existing consumer loop of topic %s because of a non-recoverable exception", topic);
+        reconnectAfterFailure();
+      } finally {
+        restClient
+            .mapAsync(kafkaRestConsumerUri, this::deleteConsumer)
+            .get(restClientTimeoutMs, TimeUnit.MILLISECONDS);
+      }
+    }
+
+    private Void seekToBeginning(Set<Integer> partitions) {
+      ListenableFuture<HttpPost> post =
+          restClient.map(
+              kafkaRestConsumerUri,
+              uri -> restClient.createPostSeekTopicFromBeginning(uri, topic, partitions));
+      restClient.map(post, restClient::execute);
+      return null;
+    }
+
+    private ListenableFuture<Set<Integer>> getTopicPartitions() {
+      HttpGet getTopic = restClient.createGetTopic(topic);
+      return restClient.mapAsync(
+          restClient.execute(getTopic, HttpStatus.SC_OK), this::getPartitions);
+    }
+
+    private ListenableFuture<ConsumerRecords<byte[], byte[]>> getRecords(URI consumerUri) {
+      HttpGet getRecords = restClient.createGetRecords(consumerUri);
+      return restClient.mapAsync(
+          restClient.execute(getRecords, HttpStatus.SC_OK), this::convertRecords);
+    }
+
+    private ListenableFuture<HttpResponse> subscribeToTopic(URI consumerUri) {
+      HttpPost post = restClient.createPostToSubscribe(consumerUri, topic);
+      return restClient.execute(post);
+    }
+
+    private ListenableFuture<?> deleteConsumer(URI consumerUri) {
+      HttpDelete delete = restClient.createDeleteToConsumer(consumerUri);
+      return restClient.execute(delete);
+    }
+
+    private ListenableFuture<URI> createConsumer(String name) {
+      HttpPost post = restClient.createPostToConsumer(name);
+      return restClient.mapAsync(restClient.execute(post, HttpStatus.SC_OK), this::getConsumerUri);
+    }
+
+    private ListenableFuture<Set<Integer>> getPartitions(HttpResponse response) {
+      try (Reader bodyReader =
+          new InputStreamReader(response.getEntity().getContent(), StandardCharsets.UTF_8)) {
+        JsonObject responseJson = gson.fromJson(bodyReader, JsonObject.class);
+        Set<Integer> partitions = extractPartitions(responseJson);
+        return Futures.immediateFuture(partitions);
+      } catch (IOException e) {
+        return Futures.immediateFailedFuture(e);
+      }
+    }
+
+    private ListenableFuture<URI> getConsumerUri(HttpResponse response) {
+      try (Reader bodyReader =
+          new InputStreamReader(response.getEntity().getContent(), StandardCharsets.UTF_8)) {
+        JsonObject responseJson = gson.fromJson(bodyReader, JsonObject.class);
+        URI consumerUri = new URI(responseJson.get("base_uri").getAsString());
+        return Futures.immediateFuture(restClient.resolveURI(consumerUri.getPath()));
+      } catch (UnsupportedOperationException | IOException | URISyntaxException e) {
+        return Futures.immediateFailedFuture(e);
+      }
+    }
+
+    private ListenableFuture<ConsumerRecords<byte[], byte[]>> convertRecords(
+        HttpResponse response) {
+      try (Reader bodyReader = new InputStreamReader(response.getEntity().getContent())) {
+        JsonArray jsonRecords = gson.fromJson(bodyReader, JsonArray.class);
+        if (jsonRecords.size() == 0) {
+          return Futures.immediateFuture(new ConsumerRecords<>(Collections.emptyMap()));
+        }
+
+        Stream<ConsumerRecord<byte[], byte[]>> jsonObjects =
+            StreamSupport.stream(jsonRecords.spliterator(), false)
+                .map(JsonElement::getAsJsonObject)
+                .map(this::jsonToConsumerRecords);
+
+        Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> records =
+            jsonObjects.collect(Collectors.groupingBy(this::jsonRecordPartition));
+        return Futures.immediateFuture(new ConsumerRecords<>(records));
+      } catch (IOException e) {
+        subscriberMetrics.incrementSubscriberFailedToConsumeMessage();
+        return Futures.immediateFailedFuture(e);
+      }
+    }
+
+    private ConsumerRecord<byte[], byte[]> jsonToConsumerRecords(JsonObject jsonRecord) {
+      return new ConsumerRecord<>(
+          jsonRecord.get("topic").getAsString(),
+          jsonRecord.get("partition").getAsInt(),
+          jsonRecord.get("offset").getAsLong(),
+          jsonRecord.get("key").toString().getBytes(),
+          jsonRecord.get("value").toString().getBytes());
+    }
+
+    private Set<Integer> extractPartitions(JsonObject jsonRecord) {
+      return StreamSupport.stream(
+              jsonRecord.get("partitions").getAsJsonArray().spliterator(), false)
+          .map(jsonElem -> jsonElem.getAsJsonObject().get("partition"))
+          .map(JsonElement::getAsInt)
+          .collect(Collectors.toSet());
+    }
+
+    private TopicPartition jsonRecordPartition(ConsumerRecord<byte[], byte[]> consumerRecord) {
+      return new TopicPartition(topic, consumerRecord.partition());
+    }
+
+    private void reconnectAfterFailure()
+        throws InterruptedException, ExecutionException, TimeoutException {
+      // Random delay with average of DELAY_RECONNECT_AFTER_FAILURE_MSEC
+      // for avoiding hammering exactly at the same interval in case of failure
+      long reconnectDelay =
+          DELAY_RECONNECT_AFTER_FAILURE_MSEC / 2
+              + new Random().nextInt(DELAY_RECONNECT_AFTER_FAILURE_MSEC);
+      Thread.sleep(reconnectDelay);
+      runReceiver();
+    }
+  }
+}

diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventSubscriber.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventSubscriber.java
index 7ef9d7b..6315dea 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventSubscriber.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventSubscriber.java

@@ -1,4 +1,4 @@
-// Copyright (C) 2019 The Android Open Source Project
+// Copyright (C) 2021 The Android Open Source Project
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -11,176 +11,39 @@
 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 // See the License for the specific language governing permissions and
 // limitations under the License.
+
 package com.googlesource.gerrit.plugins.kafka.subscribe;
 
-import static java.nio.charset.StandardCharsets.UTF_8;
-
 import com.gerritforge.gerrit.eventbroker.EventMessage;
-import com.google.common.flogger.FluentLogger;
-import com.google.gerrit.server.util.ManualRequestContext;
-import com.google.gerrit.server.util.OneOffRequestContext;
-import com.google.inject.Inject;
-import com.googlesource.gerrit.plugins.kafka.broker.ConsumerExecutor;
-import com.googlesource.gerrit.plugins.kafka.config.KafkaSubscriberProperties;
-import java.time.Duration;
-import java.util.Collections;
-import java.util.Random;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.atomic.AtomicBoolean;
-import org.apache.kafka.clients.consumer.Consumer;
-import org.apache.kafka.clients.consumer.ConsumerRecords;
-import org.apache.kafka.common.errors.WakeupException;
-import org.apache.kafka.common.serialization.Deserializer;
 
-public class KafkaEventSubscriber {
-  private static final FluentLogger logger = FluentLogger.forEnclosingClass();
-  private static final int DELAY_RECONNECT_AFTER_FAILURE_MSEC = 1000;
+/** Generic interface to a Kafka topic subscriber. */
+public interface KafkaEventSubscriber {
 
-  private final OneOffRequestContext oneOffCtx;
-  private final AtomicBoolean closed = new AtomicBoolean(false);
+  /**
+   * Subscribe to a topic and receive messages asynchronously.
+   *
+   * @param topic Kafka topic name
+   * @param messageProcessor consumer function for processing incoming messages
+   */
+  void subscribe(String topic, java.util.function.Consumer<EventMessage> messageProcessor);
 
-  private final Deserializer<EventMessage> valueDeserializer;
-  private final KafkaSubscriberProperties configuration;
-  private final ExecutorService executor;
-  private final KafkaEventSubscriberMetrics subscriberMetrics;
-  private final KafkaConsumerFactory consumerFactory;
-  private final Deserializer<byte[]> keyDeserializer;
+  /** Shutdown Kafka consumer. */
+  void shutdown();
 
-  private java.util.function.Consumer<EventMessage> messageProcessor;
-  private String topic;
-  private AtomicBoolean resetOffset = new AtomicBoolean(false);
+  /**
+   * Returns the current consumer function for the subscribed topic.
+   *
+   * @return the default topic consumer function.
+   */
+  java.util.function.Consumer<EventMessage> getMessageProcessor();
 
-  private volatile ReceiverJob receiver;
+  /**
+   * Returns the current subscribed topic name.
+   *
+   * @return Kafka topic name.
+   */
+  String getTopic();
 
-  @Inject
-  public KafkaEventSubscriber(
-      KafkaSubscriberProperties configuration,
-      KafkaConsumerFactory consumerFactory,
-      Deserializer<byte[]> keyDeserializer,
-      Deserializer<EventMessage> valueDeserializer,
-      OneOffRequestContext oneOffCtx,
-      @ConsumerExecutor ExecutorService executor,
-      KafkaEventSubscriberMetrics subscriberMetrics) {
-
-    this.configuration = configuration;
-    this.oneOffCtx = oneOffCtx;
-    this.executor = executor;
-    this.subscriberMetrics = subscriberMetrics;
-    this.consumerFactory = consumerFactory;
-    this.keyDeserializer = keyDeserializer;
-    this.valueDeserializer = valueDeserializer;
-  }
-
-  public void subscribe(String topic, java.util.function.Consumer<EventMessage> messageProcessor) {
-    this.topic = topic;
-    this.messageProcessor = messageProcessor;
-    logger.atInfo().log(
-        "Kafka consumer subscribing to topic alias [%s] for event topic [%s]", topic, topic);
-    runReceiver();
-  }
-
-  private void runReceiver() {
-    final ClassLoader previousClassLoader = Thread.currentThread().getContextClassLoader();
-    try {
-      Thread.currentThread().setContextClassLoader(KafkaEventSubscriber.class.getClassLoader());
-      Consumer<byte[], byte[]> consumer = consumerFactory.create(keyDeserializer);
-      consumer.subscribe(Collections.singleton(topic));
-      receiver = new ReceiverJob(consumer);
-      executor.execute(receiver);
-    } finally {
-      Thread.currentThread().setContextClassLoader(previousClassLoader);
-    }
-  }
-
-  public void shutdown() {
-    closed.set(true);
-    receiver.wakeup();
-  }
-
-  public java.util.function.Consumer<EventMessage> getMessageProcessor() {
-    return messageProcessor;
-  }
-
-  public String getTopic() {
-    return topic;
-  }
-
-  public void resetOffset() {
-    resetOffset.set(true);
-  }
-
-  private class ReceiverJob implements Runnable {
-    private final Consumer<byte[], byte[]> consumer;
-
-    public ReceiverJob(Consumer<byte[], byte[]> consumer) {
-      this.consumer = consumer;
-    }
-
-    public void wakeup() {
-      consumer.wakeup();
-    }
-
-    @Override
-    public void run() {
-      try {
-        consume();
-      } catch (Exception e) {
-        logger.atSevere().withCause(e).log("Consumer loop of topic %s ended", topic);
-      }
-    }
-
-    private void consume() throws InterruptedException {
-      try {
-        while (!closed.get()) {
-          if (resetOffset.getAndSet(false)) {
-            // Make sure there is an assignment for this consumer
-            while (consumer.assignment().isEmpty() && !closed.get()) {
-              logger.atInfo().log(
-                  "Resetting offset: no partitions assigned to the consumer, request assignment.");
-              consumer.poll(Duration.ofMillis(configuration.getPollingInterval()));
-            }
-            consumer.seekToBeginning(consumer.assignment());
-          }
-          ConsumerRecords<byte[], byte[]> consumerRecords =
-              consumer.poll(Duration.ofMillis(configuration.getPollingInterval()));
-          consumerRecords.forEach(
-              consumerRecord -> {
-                try (ManualRequestContext ctx = oneOffCtx.open()) {
-                  EventMessage event =
-                      valueDeserializer.deserialize(consumerRecord.topic(), consumerRecord.value());
-                  messageProcessor.accept(event);
-                } catch (Exception e) {
-                  logger.atSevere().withCause(e).log(
-                      "Malformed event '%s': [Exception: %s]",
-                      new String(consumerRecord.value(), UTF_8));
-                  subscriberMetrics.incrementSubscriberFailedToConsumeMessage();
-                }
-              });
-        }
-      } catch (WakeupException e) {
-        // Ignore exception if closing
-        if (!closed.get()) {
-          logger.atSevere().withCause(e).log("Consumer loop of topic %s interrupted", topic);
-          reconnectAfterFailure();
-        }
-      } catch (Exception e) {
-        subscriberMetrics.incrementSubscriberFailedToPollMessages();
-        logger.atSevere().withCause(e).log(
-            "Existing consumer loop of topic %s because of a non-recoverable exception", topic);
-        reconnectAfterFailure();
-      } finally {
-        consumer.close();
-      }
-    }
-
-    private void reconnectAfterFailure() throws InterruptedException {
-      // Random delay with average of DELAY_RECONNECT_AFTER_FAILURE_MSEC
-      // for avoiding hammering exactly at the same interval in case of failure
-      long reconnectDelay =
-          DELAY_RECONNECT_AFTER_FAILURE_MSEC / 2
-              + new Random().nextInt(DELAY_RECONNECT_AFTER_FAILURE_MSEC);
-      Thread.sleep(reconnectDelay);
-      runReceiver();
-    }
-  }
+  /** Reset the offset for reading incoming Kafka messages of the topic. */
+  void resetOffset();
 }

diff --git a/src/main/resources/Documentation/config.md b/src/main/resources/Documentation/config.md
index 727b7e5..afe775e 100644
--- a/src/main/resources/Documentation/config.md
+++ b/src/main/resources/Documentation/config.md

@@ -36,14 +36,55 @@
 Additional properties
 ---------------------
 
+`plugin.@PLUGIN@.clientType`
+:	Client stack for connecting to Kafka broker:
+    - `NATIVE` for using the Kafka client to connect to the broker directory
+    - `REST` for using a simple HTTP client to connect to
+      [Confluent REST-API Proxy](https://docs.confluent.io/platform/current/kafka-rest/index.html).
+      **NOTE**: `plugin.@PLUGIN@.restApiUri` is mandatory when using a `REST` client type.
+	Default: `NATIVE`
+
 `plugin.@PLUGIN@.groupId`
 :	Kafka consumer group for receiving messages.
 	Default: Gerrit instance-id
 
+`plugin.@PLUGIN@.httpWireLog`
+:	Enable the HTTP wire protocol logging in error_log for all the communication with
+	the [Confluent REST-API Proxy](https://docs.confluent.io/platform/current/kafka-rest/index.html).
+	**NOTE**: when `plugin.@PLUGIN@.restApiUri` is unset or set to `NATIVE`, this setting is ignored.
+	Default: false
+
 `plugin.@PLUGIN@.pollingIntervalMs`
 :	Polling interval in msec for receiving messages from Kafka topic subscription.
 	Default: 1000
 
+`plugin.@PLUGIN@.restApiUri`
+:	URL of the
+	[Confluent REST-API Proxy](https://docs.confluent.io/platform/current/kafka-rest/index.html)
+	for sending/receiving messages through REST-API instead of using the native Kafka client.
+	**NOTE**: when `plugin.@PLUGIN@.restApiUri` is unset or set to `NATIVE`, this setting is ignored.
+	Default: unset
+
+`plugin.@PLUGIN@.restApiThreads`
+:	Maximum number of concurrent client calls to the
+	[Confluent REST-API Proxy](https://docs.confluent.io/platform/current/kafka-rest/index.html)
+	for sending/receiving messages.
+	**NOTE**: when `plugin.@PLUGIN@.restApiUri` is unset or set to `NATIVE`, this setting is ignored.
+	Default: 10
+
+`plugin.@PLUGIN@.restApiTimeout`
+:	Maximum time to wait for a client call to
+	[Confluent REST-API Proxy](https://docs.confluent.io/platform/current/kafka-rest/index.html)
+	to complete. This setting is also applied as TCP socket connection and read/write timeout
+	for the outgoing HTTP calls.
+	The value is expressed using the `N unit` format of all other Gerrit time expressions, using
+	one of the following units:
+	- s, sec, second, seconds
+	- m, min, minute, minutes
+	- h, hr, hour, hours
+	**NOTE**: when `plugin.@PLUGIN@.restApiUri` is unset or set to `NATIVE`, this setting is ignored.
+	Default: 60 sec
+
 `plugin.@PLUGIN@.sendAsync`
 :	Send messages to Kafka asynchronously, detaching the calling process from the
 	acknowledge of the message being sent.

diff --git a/src/test/java/com/googlesource/gerrit/plugins/kafka/EventConsumerIT.java b/src/test/java/com/googlesource/gerrit/plugins/kafka/EventConsumerIT.java
index 118a868..85945c0 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/kafka/EventConsumerIT.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/kafka/EventConsumerIT.java

@@ -53,14 +53,14 @@
 @NoHttpd
 @TestPlugin(name = "events-kafka", sysModule = "com.googlesource.gerrit.plugins.kafka.Module")
 public class EventConsumerIT extends LightweightPluginDaemonTest {
-  static final long KAFKA_POLL_TIMEOUT = 10000L;
+  static final Duration KAFKA_POLL_TIMEOUT = Duration.ofSeconds(10);
 
   private KafkaContainer kafka;
 
   @Override
   public void setUpTestPlugin() throws Exception {
     try {
-      kafka = new KafkaContainer();
+      kafka = KafkaContainerProvider.get();
       kafka.start();
 
       System.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers());
@@ -146,7 +146,7 @@
             new EventMessage.Header(UUID.randomUUID(), UUID.randomUUID()),
             new ProjectCreatedEvent());
 
-    Duration WAIT_FOR_POLL_TIMEOUT = Duration.ofMillis(1000);
+    Duration WAIT_FOR_POLL_TIMEOUT = Duration.ofSeconds(30);
 
     List<EventMessage> receivedEvents = new ArrayList<>();
 

diff --git a/src/test/java/com/googlesource/gerrit/plugins/kafka/KafkaContainerProvider.java b/src/test/java/com/googlesource/gerrit/plugins/kafka/KafkaContainerProvider.java
new file mode 100644
index 0000000..859d7ee
--- /dev/null
+++ b/src/test/java/com/googlesource/gerrit/plugins/kafka/KafkaContainerProvider.java

@@ -0,0 +1,50 @@
+// Copyright (C) 2021 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.googlesource.gerrit.plugins.kafka;
+
+import java.util.Map;
+import org.junit.Ignore;
+import org.testcontainers.containers.KafkaContainer;
+import org.testcontainers.containers.Network;
+import org.testcontainers.utility.DockerImageName;
+
+@Ignore
+public class KafkaContainerProvider {
+  public static int KAFKA_PORT_INTERNAL = KafkaContainer.KAFKA_PORT + 1;
+  private static final String KAFKA_IMAGE_NAME = "confluentinc/cp-kafka";
+  private static final String KAFKA_IMAGE_TAG = "5.4.3";
+
+  public static KafkaContainer get() {
+    KafkaContainer kafkaContainer =
+        new KafkaContainer(DockerImageName.parse(KAFKA_IMAGE_NAME).withTag(KAFKA_IMAGE_TAG)) {
+
+          @Override
+          public String getBootstrapServers() {
+            return String.format(
+                    "INTERNAL://%s:%s,", getNetworkAliases().get(0), KAFKA_PORT_INTERNAL)
+                + super.getBootstrapServers();
+          }
+        };
+
+    Map<String, String> kafkaEnv = kafkaContainer.getEnvMap();
+    String kafkaListeners = kafkaEnv.get("KAFKA_LISTENERS");
+    String kafkaProtocolMap = kafkaEnv.get("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP");
+
+    return kafkaContainer
+        .withNetwork(Network.newNetwork())
+        .withEnv("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP", kafkaProtocolMap + ",INTERNAL:PLAINTEXT")
+        .withEnv("KAFKA_LISTENERS", kafkaListeners + ",INTERNAL://0.0.0.0:" + KAFKA_PORT_INTERNAL);
+  }
+}

diff --git a/src/test/java/com/googlesource/gerrit/plugins/kafka/KafkaRestContainer.java b/src/test/java/com/googlesource/gerrit/plugins/kafka/KafkaRestContainer.java
new file mode 100644
index 0000000..dbcc455
--- /dev/null
+++ b/src/test/java/com/googlesource/gerrit/plugins/kafka/KafkaRestContainer.java

@@ -0,0 +1,61 @@
+// Copyright (C) 2021 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.googlesource.gerrit.plugins.kafka;
+
+import java.net.URI;
+import java.net.URISyntaxException;
+import org.junit.Ignore;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.KafkaContainer;
+import org.testcontainers.utility.DockerImageName;
+
+@Ignore
+public class KafkaRestContainer extends GenericContainer<KafkaRestContainer> {
+
+  private static final String KAFKA_REST_PROXY_HOSTNAME = "restproxy";
+
+  public static final int KAFKA_REST_PORT = 8082;
+
+  public KafkaRestContainer(KafkaContainer kafkaContainer) {
+    super(restProxyImageFor(kafkaContainer));
+
+    withNetwork(kafkaContainer.getNetwork());
+
+    withExposedPorts(KAFKA_REST_PORT);
+    String bootstrapServers =
+        String.format(
+            "PLAINTEXT://%s:%s",
+            kafkaContainer.getNetworkAliases().get(0), KafkaContainerProvider.KAFKA_PORT_INTERNAL);
+    withEnv("KAFKA_REST_BOOTSTRAP_SERVERS", bootstrapServers);
+    withEnv("KAFKA_REST_LISTENERS", "http://0.0.0.0:" + KAFKA_REST_PORT);
+    withEnv("KAFKA_REST_CLIENT_SECURITY_PROTOCOL", "PLAINTEXT");
+    withEnv("KAFKA_REST_HOST_NAME", KAFKA_REST_PROXY_HOSTNAME);
+    withCreateContainerCmdModifier(cmd -> cmd.withHostName(KAFKA_REST_PROXY_HOSTNAME));
+  }
+
+  private static DockerImageName restProxyImageFor(KafkaContainer kafkaContainer) {
+    String[] kafkaImageNameParts = kafkaContainer.getDockerImageName().split(":");
+    return DockerImageName.parse(kafkaImageNameParts[0] + "-rest").withTag(kafkaImageNameParts[1]);
+  }
+
+  public URI getApiURI() {
+    try {
+      return new URI(
+          String.format("http://%s:%d", getContainerIpAddress(), getMappedPort(KAFKA_REST_PORT)));
+    } catch (URISyntaxException e) {
+      throw new IllegalArgumentException("Invalid Kafka API URI", e);
+    }
+  }
+}

diff --git a/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerApiTest.java b/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerApiTest.java
index 48350f9..2a2d6d4 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerApiTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerApiTest.java

@@ -33,20 +33,25 @@
 import com.google.inject.Scopes;
 import com.google.inject.Singleton;
 import com.google.inject.TypeLiteral;
+import com.googlesource.gerrit.plugins.kafka.KafkaContainerProvider;
+import com.googlesource.gerrit.plugins.kafka.KafkaRestContainer;
 import com.googlesource.gerrit.plugins.kafka.config.KafkaProperties;
+import com.googlesource.gerrit.plugins.kafka.config.KafkaProperties.ClientType;
 import com.googlesource.gerrit.plugins.kafka.config.KafkaSubscriberProperties;
 import com.googlesource.gerrit.plugins.kafka.session.KafkaProducerProvider;
 import com.googlesource.gerrit.plugins.kafka.session.KafkaSession;
+import java.net.URI;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.UUID;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
-import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.junit.After;
 import org.junit.AfterClass;
+import org.junit.Before;
 import org.junit.BeforeClass;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -56,19 +61,23 @@
 
 @RunWith(MockitoJUnitRunner.class)
 public class KafkaBrokerApiTest {
-  private static KafkaContainer kafka;
 
-  private static final int TEST_NUM_SUBSCRIBERS = 1;
-  private static final String TEST_GROUP_ID = KafkaBrokerApiTest.class.getName();
-  private static final int TEST_POLLING_INTERVAL_MSEC = 100;
+  static KafkaContainer kafka;
+  static KafkaRestContainer kafkaRest;
+
+  static final int TEST_NUM_SUBSCRIBERS = 1;
+  static final String TEST_GROUP_ID = KafkaBrokerApiTest.class.getName();
+  static final int TEST_POLLING_INTERVAL_MSEC = 100;
   private static final int TEST_THREAD_POOL_SIZE = 10;
   private static final UUID TEST_INSTANCE_ID = UUID.randomUUID();
-  private static final TimeUnit TEST_TIMOUT_UNIT = TimeUnit.SECONDS;
+  private static final TimeUnit TEST_TIMEOUT_UNIT = TimeUnit.SECONDS;
   private static final int TEST_TIMEOUT = 30;
+  private static final int TEST_WAIT_FOR_MORE_MESSAGES_TIMEOUT = 5;
 
   private Injector injector;
   private KafkaSession session;
   private Gson gson;
+  protected ClientType clientType;
 
   public static class TestWorkQueue extends WorkQueue {
 
@@ -94,22 +103,30 @@
 
       bind(KafkaProperties.class).toInstance(kafkaProperties);
       bind(KafkaSession.class).in(Scopes.SINGLETON);
-      KafkaSubscriberProperties kafkaSubscriberProperties =
-          new KafkaSubscriberProperties(
-              TEST_POLLING_INTERVAL_MSEC, TEST_GROUP_ID, TEST_NUM_SUBSCRIBERS);
-      bind(KafkaSubscriberProperties.class).toInstance(kafkaSubscriberProperties);
-      bind(new TypeLiteral<KafkaProducer<String, String>>() {})
-          .toProvider(KafkaProducerProvider.class);
+
+      bindKafkaClientImpl();
 
       bind(WorkQueue.class).to(TestWorkQueue.class);
     }
+
+    protected void bindKafkaClientImpl() {
+      bind(new TypeLiteral<Producer<String, String>>() {}).toProvider(KafkaProducerProvider.class);
+      KafkaSubscriberProperties kafkaSubscriberProperties =
+          new KafkaSubscriberProperties(
+              TEST_POLLING_INTERVAL_MSEC, TEST_GROUP_ID, TEST_NUM_SUBSCRIBERS, ClientType.NATIVE);
+      bind(KafkaSubscriberProperties.class).toInstance(kafkaSubscriberProperties);
+    }
   }
 
   public static class TestConsumer implements Consumer<EventMessage> {
     public final List<EventMessage> messages = new ArrayList<>();
-    private final CountDownLatch lock;
+    private CountDownLatch lock;
 
     public TestConsumer(int numMessagesExpected) {
+      resetExpectedMessages(numMessagesExpected);
+    }
+
+    public void resetExpectedMessages(int numMessagesExpected) {
       lock = new CountDownLatch(numMessagesExpected);
     }
 
@@ -120,8 +137,12 @@
     }
 
     public boolean await() {
+      return await(TEST_TIMEOUT, TEST_TIMEOUT_UNIT);
+    }
+
+    public boolean await(long timeout, TimeUnit unit) {
       try {
-        return lock.await(TEST_TIMEOUT, TEST_TIMOUT_UNIT);
+        return lock.await(timeout, unit);
       } catch (InterruptedException e) {
         return false;
       }
@@ -137,11 +158,18 @@
 
   @BeforeClass
   public static void beforeClass() throws Exception {
-    kafka = new KafkaContainer();
+    kafka = KafkaContainerProvider.get();
     kafka.start();
+    kafkaRest = new KafkaRestContainer(kafka);
+    kafkaRest.start();
     System.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers());
   }
 
+  @Before
+  public void setup() {
+    clientType = ClientType.NATIVE;
+  }
+
   @AfterClass
   public static void afterClass() {
     if (kafka != null) {
@@ -149,8 +177,12 @@
     }
   }
 
+  protected TestModule newTestModule(KafkaProperties kafkaProperties) {
+    return new TestModule(kafkaProperties);
+  }
+
   public void connectToKafka(KafkaProperties kafkaProperties) {
-    Injector baseInjector = Guice.createInjector(new TestModule(kafkaProperties));
+    Injector baseInjector = Guice.createInjector(newTestModule(kafkaProperties));
     WorkQueue testWorkQueue = baseInjector.getInstance(WorkQueue.class);
     KafkaSubscriberProperties kafkaSubscriberProperties =
         baseInjector.getInstance(KafkaSubscriberProperties.class);
@@ -172,7 +204,7 @@
 
   @Test
   public void shouldSendSyncAndReceiveToTopic() {
-    connectToKafka(new KafkaProperties(false));
+    connectToKafka(new KafkaProperties(false, clientType, getKafkaRestApiURI()));
     KafkaBrokerApi kafkaBrokerApi = injector.getInstance(KafkaBrokerApi.class);
     String testTopic = "test_topic_sync";
     TestConsumer testConsumer = new TestConsumer(1);
@@ -184,11 +216,13 @@
     assertThat(testConsumer.await()).isTrue();
     assertThat(testConsumer.messages).hasSize(1);
     assertThat(gson.toJson(testConsumer.messages.get(0))).isEqualTo(gson.toJson(testEventMessage));
+
+    assertNoMoreExpectedMessages(testConsumer);
   }
 
   @Test
   public void shouldSendAsyncAndReceiveToTopic() {
-    connectToKafka(new KafkaProperties(true));
+    connectToKafka(new KafkaProperties(true, clientType, getKafkaRestApiURI()));
     KafkaBrokerApi kafkaBrokerApi = injector.getInstance(KafkaBrokerApi.class);
     String testTopic = "test_topic_async";
     TestConsumer testConsumer = new TestConsumer(1);
@@ -200,5 +234,39 @@
     assertThat(testConsumer.await()).isTrue();
     assertThat(testConsumer.messages).hasSize(1);
     assertThat(gson.toJson(testConsumer.messages.get(0))).isEqualTo(gson.toJson(testEventMessage));
+
+    assertNoMoreExpectedMessages(testConsumer);
+  }
+
+  @Test
+  public void shouldSendToTopicAndResetOffset() {
+    connectToKafka(new KafkaProperties(false, clientType, getKafkaRestApiURI()));
+    KafkaBrokerApi kafkaBrokerApi = injector.getInstance(KafkaBrokerApi.class);
+    String testTopic = "test_topic_reset";
+    TestConsumer testConsumer = new TestConsumer(1);
+    EventMessage testEventMessage = new EventMessage(new TestHeader(), new ProjectCreatedEvent());
+
+    kafkaBrokerApi.receiveAsync(testTopic, testConsumer);
+    kafkaBrokerApi.send(testTopic, testEventMessage);
+
+    assertThat(testConsumer.await()).isTrue();
+    assertThat(testConsumer.messages).hasSize(1);
+    assertThat(gson.toJson(testConsumer.messages.get(0))).isEqualTo(gson.toJson(testEventMessage));
+
+    kafkaBrokerApi.replayAllEvents(testTopic);
+
+    assertThat(testConsumer.await()).isTrue();
+    assertThat(testConsumer.messages).hasSize(1);
+    assertThat(gson.toJson(testConsumer.messages.get(0))).isEqualTo(gson.toJson(testEventMessage));
+  }
+
+  protected URI getKafkaRestApiURI() {
+    return null;
+  }
+
+  private void assertNoMoreExpectedMessages(TestConsumer testConsumer) {
+    testConsumer.resetExpectedMessages(1);
+    assertThat(testConsumer.await(TEST_WAIT_FOR_MORE_MESSAGES_TIMEOUT, TEST_TIMEOUT_UNIT))
+        .isFalse();
   }
 }

diff --git a/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerRestApiTest.java b/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerRestApiTest.java
new file mode 100644
index 0000000..d5f3c64
--- /dev/null
+++ b/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerRestApiTest.java

@@ -0,0 +1,74 @@
+// Copyright (C) 2021 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.googlesource.gerrit.plugins.kafka.api;
+
+import com.google.inject.TypeLiteral;
+import com.google.inject.assistedinject.FactoryModuleBuilder;
+import com.googlesource.gerrit.plugins.kafka.config.KafkaProperties;
+import com.googlesource.gerrit.plugins.kafka.config.KafkaProperties.ClientType;
+import com.googlesource.gerrit.plugins.kafka.config.KafkaSubscriberProperties;
+import com.googlesource.gerrit.plugins.kafka.publish.KafkaRestProducer;
+import com.googlesource.gerrit.plugins.kafka.rest.FutureExecutor;
+import com.googlesource.gerrit.plugins.kafka.rest.HttpHostProxy;
+import com.googlesource.gerrit.plugins.kafka.rest.KafkaRestClient;
+import java.net.URI;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import org.apache.kafka.clients.producer.Producer;
+import org.junit.Before;
+import org.junit.runner.RunWith;
+import org.mockito.junit.MockitoJUnitRunner;
+
+@RunWith(MockitoJUnitRunner.class)
+public class KafkaBrokerRestApiTest extends KafkaBrokerApiTest {
+
+  @Override
+  @Before
+  public void setup() {
+    clientType = ClientType.REST;
+  }
+
+  @Override
+  protected TestModule newTestModule(KafkaProperties kafkaProperties) {
+    return new TestModule(kafkaProperties) {
+
+      @Override
+      protected void bindKafkaClientImpl() {
+        bind(new TypeLiteral<Producer<String, String>>() {}).to(KafkaRestProducer.class);
+        bind(ExecutorService.class)
+            .annotatedWith(FutureExecutor.class)
+            .toInstance(Executors.newCachedThreadPool());
+
+        KafkaSubscriberProperties kafkaSubscriberProperties =
+            new KafkaSubscriberProperties(
+                TEST_POLLING_INTERVAL_MSEC,
+                TEST_GROUP_ID,
+                TEST_NUM_SUBSCRIBERS,
+                ClientType.REST,
+                kafkaRest.getApiURI());
+        bind(KafkaSubscriberProperties.class).toInstance(kafkaSubscriberProperties);
+
+        bind(HttpHostProxy.class).toInstance(new HttpHostProxy(null, null, null));
+
+        install(new FactoryModuleBuilder().build(KafkaRestClient.Factory.class));
+      }
+    };
+  }
+
+  @Override
+  protected URI getKafkaRestApiURI() {
+    return kafkaRest.getApiURI();
+  }
+}

diff --git a/src/test/java/com/googlesource/gerrit/plugins/kafka/publish/KafkaSessionTest.java b/src/test/java/com/googlesource/gerrit/plugins/kafka/publish/KafkaSessionTest.java
index 5aa9ca8..59e7ca2 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/kafka/publish/KafkaSessionTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/kafka/publish/KafkaSessionTest.java

@@ -14,6 +14,7 @@
 
 package com.googlesource.gerrit.plugins.kafka.publish;
 
+import static com.google.common.truth.Truth.assertThat;
 import static org.mockito.Mockito.any;
 import static org.mockito.Mockito.only;
 import static org.mockito.Mockito.verify;
@@ -21,10 +22,11 @@
 
 import com.google.common.util.concurrent.Futures;
 import com.googlesource.gerrit.plugins.kafka.config.KafkaProperties;
+import com.googlesource.gerrit.plugins.kafka.config.KafkaProperties.ClientType;
 import com.googlesource.gerrit.plugins.kafka.session.KafkaProducerProvider;
 import com.googlesource.gerrit.plugins.kafka.session.KafkaSession;
 import org.apache.kafka.clients.producer.Callback;
-import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.RecordMetadata;
 import org.apache.kafka.common.TopicPartition;
 import org.junit.Before;
@@ -38,7 +40,7 @@
 @RunWith(MockitoJUnitRunner.class)
 public class KafkaSessionTest {
   KafkaSession objectUnderTest;
-  @Mock KafkaProducer<String, String> kafkaProducer;
+  @Mock Producer<String, String> kafkaProducer;
   @Mock KafkaProducerProvider producerProvider;
   @Mock KafkaProperties properties;
   @Mock KafkaEventsPublisherMetrics publisherMetrics;
@@ -52,17 +54,19 @@
   public void setUp() {
     when(producerProvider.get()).thenReturn(kafkaProducer);
     when(properties.getTopic()).thenReturn(topic);
+    when(properties.getProperty("bootstrap.servers")).thenReturn("localhost:9092");
+    when(properties.getClientType()).thenReturn(ClientType.NATIVE);
 
     recordMetadata = new RecordMetadata(new TopicPartition(topic, 0), 0L, 0L, 0L, 0L, 0, 0);
 
     objectUnderTest = new KafkaSession(producerProvider, properties, publisherMetrics);
-    objectUnderTest.connect();
   }
 
   @Test
   public void shouldIncrementBrokerMetricCounterWhenMessagePublishedInSyncMode() {
     when(properties.isSendAsync()).thenReturn(false);
     when(kafkaProducer.send(any())).thenReturn(Futures.immediateFuture(recordMetadata));
+    objectUnderTest.connect();
     objectUnderTest.publish(message);
     verify(publisherMetrics, only()).incrementBrokerPublishedMessage();
   }
@@ -71,6 +75,7 @@
   public void shouldIncrementBrokerFailedMetricCounterWhenMessagePublishingFailedInSyncMode() {
     when(properties.isSendAsync()).thenReturn(false);
     when(kafkaProducer.send(any())).thenReturn(Futures.immediateFailedFuture(new Exception()));
+    objectUnderTest.connect();
     objectUnderTest.publish(message);
     verify(publisherMetrics, only()).incrementBrokerFailedToPublishMessage();
   }
@@ -80,6 +85,7 @@
     when(properties.isSendAsync()).thenReturn(false);
     when(kafkaProducer.send(any())).thenThrow(new RuntimeException("Unexpected runtime exception"));
     try {
+      objectUnderTest.connect();
       objectUnderTest.publish(message);
     } catch (RuntimeException e) {
       // expected
@@ -92,6 +98,7 @@
     when(properties.isSendAsync()).thenReturn(true);
     when(kafkaProducer.send(any(), any())).thenReturn(Futures.immediateFuture(recordMetadata));
 
+    objectUnderTest.connect();
     objectUnderTest.publish(message);
 
     verify(kafkaProducer).send(any(), callbackCaptor.capture());
@@ -105,6 +112,7 @@
     when(kafkaProducer.send(any(), any()))
         .thenReturn(Futures.immediateFailedFuture(new Exception()));
 
+    objectUnderTest.connect();
     objectUnderTest.publish(message);
 
     verify(kafkaProducer).send(any(), callbackCaptor.capture());
@@ -118,10 +126,18 @@
     when(kafkaProducer.send(any(), any()))
         .thenThrow(new RuntimeException("Unexpected runtime exception"));
     try {
+      objectUnderTest.connect();
       objectUnderTest.publish(message);
     } catch (RuntimeException e) {
       // expected
     }
     verify(publisherMetrics, only()).incrementBrokerFailedToPublishMessage();
   }
+
+  @Test
+  public void shouldNotConnectKafkaSessionWhenBoostrapServersAreNotSet() {
+    when(properties.getProperty("bootstrap.servers")).thenReturn(null);
+    objectUnderTest.connect();
+    assertThat(objectUnderTest.isOpen()).isFalse();
+  }
 }

diff --git a/src/test/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventDeserializerTest.java b/src/test/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventDeserializerTest.java
index e456a2a..4074919 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventDeserializerTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventDeserializerTest.java

@@ -34,10 +34,10 @@
   }
 
   @Test
-  public void kafkaEventDeserializerShouldParseAKafkaEvent() {
+  public void kafkaEventDeserializerShouldParseAKafkaEventMessage() {
     final UUID eventId = UUID.randomUUID();
     final String eventType = "event-type";
-    final UUID sourceInstanceId = UUID.randomUUID();
+    final String sourceInstanceId = UUID.randomUUID().toString();
     final long eventCreatedOn = 10L;
     final String eventJson =
         String.format(
@@ -52,6 +52,23 @@
     assertThat(event.getHeader().sourceInstanceId).isEqualTo(sourceInstanceId);
   }
 
+  @Test
+  public void kafkaEventDeserializerShouldParseKafkaEvent() {
+    final String eventJson = "{ \"type\": \"project-created\", \"instanceId\":\"instance-id\" }";
+    final EventMessage event = deserializer.deserialize("ignored", eventJson.getBytes(UTF_8));
+
+    assertThat(event.getHeader().sourceInstanceId).isEqualTo("instance-id");
+  }
+
+  @Test
+  public void kafkaEventDeserializerShouldParseKafkaEventWithHeaderAndBodyProjectName() {
+    final String eventJson =
+        "{\"projectName\":\"header_body_parser_project\",\"type\":\"project-created\", \"instanceId\":\"instance-id\"}";
+    final EventMessage event = deserializer.deserialize("ignored", eventJson.getBytes(UTF_8));
+
+    assertThat(event.getHeader().sourceInstanceId).isEqualTo("instance-id");
+  }
+
   @Test(expected = RuntimeException.class)
   public void kafkaEventDeserializerShouldFailForInvalidJson() {
     deserializer.deserialize("ignored", "this is not a JSON string".getBytes(UTF_8));