diff --git a/BUILD b/BUILD
index d9f938b..8889c80 100644
--- a/BUILD
+++ b/BUILD
@@ -28,6 +28,7 @@
 junit_tests(
     name = "events_kafka_tests",
     srcs = glob(["src/test/java/**/*.java"]),
+    resources = glob(["src/test/resources/**/*"]),
     tags = ["events-kafka"],
     timeout = "long",
     deps = [
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 62330e3..1b29748 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
@@ -40,6 +40,8 @@
   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_USERNAME = "restApiUsername";
+  private static final String PROPERTY_REST_API_PASSWORD = "restApiPassword";
   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";
@@ -64,6 +66,8 @@
   private final boolean sendAsync;
   private final ClientType clientType;
   private final String restApiUriString;
+  private final String restApiUsername;
+  private final String restApiPassword;
   private final boolean httpWireLog;
   private final Duration restApiTimeout;
   private final int restApiThreads;
@@ -86,6 +90,12 @@
           throw new IllegalArgumentException("Missing REST API URI in Kafka properties");
         }
 
+        restApiUsername = fromGerritConfig.getString(PROPERTY_REST_API_USERNAME);
+        restApiPassword = fromGerritConfig.getString(PROPERTY_REST_API_PASSWORD);
+        if (!Strings.isNullOrEmpty(restApiUsername) && Strings.isNullOrEmpty(restApiPassword)) {
+          throw new IllegalArgumentException("Missing REST API password in kafka properties");
+        }
+
         httpWireLog = fromGerritConfig.getBoolean(PROPERTY_HTTP_WIRE_LOG, DEFAULT_HTTP_WIRE_LOG);
         restApiTimeout =
             Duration.ofMillis(
@@ -99,6 +109,8 @@
       case NATIVE:
       default:
         restApiUriString = null;
+        restApiUsername = null;
+        restApiPassword = null;
         httpWireLog = false;
         restApiTimeout = null;
         restApiThreads = 0;
@@ -111,7 +123,11 @@
 
   @VisibleForTesting
   public KafkaProperties(
-      boolean sendAsync, ClientType clientType, @Nullable String restApiUriString) {
+      boolean sendAsync,
+      ClientType clientType,
+      @Nullable String restApiUriString,
+      @Nullable String restApiUsername,
+      @Nullable String restApiPassword) {
     super();
     setDefaults();
     topic = DEFAULT_STREAM_EVENTS_TOPIC_NAME;
@@ -122,6 +138,8 @@
     this.httpWireLog = false;
     restApiTimeout = DEFAULT_REST_API_TIMEOUT;
     restApiThreads = DEFAULT_REST_API_THREADS;
+    this.restApiUsername = restApiUsername;
+    this.restApiPassword = restApiPassword;
   }
 
   private void setDefaults() {
@@ -176,6 +194,14 @@
     return getRestApiUri("");
   }
 
+  public String getRestApiUsername() {
+    return restApiUsername;
+  }
+
+  public String getRestApiPassword() {
+    return restApiPassword;
+  }
+
   public URI getRestApiUri(String kafkaRestId) throws URISyntaxException {
     return new URI(restApiUriString.replace(REST_API_URI_ID_PLACEHOLDER, kafkaRestId));
   }
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 9b4ef8f..7e6a49c 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
@@ -45,7 +45,7 @@
   @VisibleForTesting
   public KafkaSubscriberProperties(
       int pollingInterval, String groupId, int numberOfSubscribers, ClientType clientType) {
-    this(pollingInterval, groupId, numberOfSubscribers, clientType, null);
+    this(pollingInterval, groupId, numberOfSubscribers, clientType, null, null, null);
   }
 
   @VisibleForTesting
@@ -54,8 +54,10 @@
       String groupId,
       int numberOfSubscribers,
       ClientType clientType,
-      String restApiUriString) {
-    super(true, clientType, restApiUriString);
+      String restApiUriString,
+      String restApiUsername,
+      String restApiPassword) {
+    super(true, clientType, restApiUriString, restApiUsername, restApiPassword);
     this.pollingInterval = pollingInterval;
     this.groupId = groupId;
     this.numberOfSubscribers = numberOfSubscribers;
diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/rest/HttpAsyncClientBuilderFactory.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/rest/HttpAsyncClientBuilderFactory.java
new file mode 100644
index 0000000..ec50bf6
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/rest/HttpAsyncClientBuilderFactory.java
@@ -0,0 +1,50 @@
+// Copyright (C) 2011 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.inject.Inject;
+import com.googlesource.gerrit.plugins.kafka.config.KafkaProperties;
+import java.net.URISyntaxException;
+import org.apache.http.auth.AuthScope;
+import org.apache.http.auth.UsernamePasswordCredentials;
+import org.apache.http.client.CredentialsProvider;
+import org.apache.http.impl.client.BasicCredentialsProvider;
+import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
+import org.apache.http.impl.nio.client.HttpAsyncClients;
+
+/** Looks up a remote's password in secure.config. */
+public class HttpAsyncClientBuilderFactory {
+  private final KafkaProperties config;
+
+  @Inject
+  public HttpAsyncClientBuilderFactory(KafkaProperties config) {
+    this.config = config;
+  }
+
+  public HttpAsyncClientBuilder create() throws URISyntaxException {
+    String user = config.getRestApiUsername();
+    String pass = config.getRestApiPassword();
+    HttpAsyncClientBuilder httpAsyncClientBuilder = HttpAsyncClients.custom();
+    if (!Strings.isNullOrEmpty(user) && !Strings.isNullOrEmpty(pass)) {
+      CredentialsProvider credsProvider = new BasicCredentialsProvider();
+      credsProvider.setCredentials(
+          new AuthScope(config.getRestApiUri().getHost(), config.getRestApiUri().getPort()),
+          new UsernamePasswordCredentials(user, pass));
+      httpAsyncClientBuilder.setDefaultCredentialsProvider(credsProvider);
+    }
+    return httpAsyncClientBuilder;
+  }
+}
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
index d06b427..047a96e 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/kafka/rest/KafkaRestClient.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/rest/KafkaRestClient.java
@@ -50,7 +50,6 @@
 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.Appender;
 import org.apache.log4j.AppenderSkeleton;
 import org.apache.log4j.Level;
@@ -78,9 +77,11 @@
   public KafkaRestClient(
       HttpHostProxy httpHostProxy,
       @FutureExecutor ExecutorService executor,
-      @Assisted KafkaProperties configuration) {
+      HttpAsyncClientBuilderFactory credentialsFactory,
+      @Assisted KafkaProperties configuration)
+      throws URISyntaxException {
     proxy = httpHostProxy;
-    httpclient = proxy.apply(HttpAsyncClients.custom()).build();
+    httpclient = proxy.apply(credentialsFactory.create()).build();
     httpclient.start();
     this.configuration = configuration;
     kafkaRestApiTimeoutMsec = (int) configuration.getRestApiTimeout().toMillis();
diff --git a/src/main/resources/Documentation/config.md b/src/main/resources/Documentation/config.md
index 24bdc8c..bfac4c5 100644
--- a/src/main/resources/Documentation/config.md
+++ b/src/main/resources/Documentation/config.md
@@ -93,3 +93,19 @@
 :	Send messages to Kafka asynchronously, detaching the calling process from the
 	acknowledge of the message being sent.
 	Default: true
+
+secure.config
+--------------------
+
+`plugin.@PLUGIN@.restApiUsername`
+:	Username used for the authentication to 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@.restApiPassword`
+:	Password used for the authentication to 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: unset
diff --git a/src/test/java/com/googlesource/gerrit/plugins/kafka/KafkaRestContainer.java b/src/test/java/com/googlesource/gerrit/plugins/kafka/KafkaRestContainer.java
index 06c6801..5cbf585 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/kafka/KafkaRestContainer.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/kafka/KafkaRestContainer.java
@@ -20,6 +20,7 @@
 import java.net.URISyntaxException;
 import java.util.Map;
 import org.junit.Ignore;
+import org.testcontainers.containers.BindMode;
 import org.testcontainers.containers.GenericContainer;
 import org.testcontainers.containers.KafkaContainer;
 import org.testcontainers.utility.DockerImageName;
@@ -31,11 +32,12 @@
   public static final int KAFKA_REST_PORT = 8082;
   private final String kafkaRestHostname;
 
-  public KafkaRestContainer(KafkaContainer kafkaContainer) {
-    this(kafkaContainer, null);
+  public KafkaRestContainer(KafkaContainer kafkaContainer, Boolean enableAuthentication) {
+    this(kafkaContainer, null, enableAuthentication);
   }
 
-  public KafkaRestContainer(KafkaContainer kafkaContainer, String kafkaRestId) {
+  public KafkaRestContainer(
+      KafkaContainer kafkaContainer, String kafkaRestId, Boolean enableAuthentication) {
     super(restProxyImageFor(kafkaContainer));
 
     kafkaRestHostname = KAFKA_REST_PROXY_HOSTNAME + Strings.nullToEmpty(kafkaRestId);
@@ -54,6 +56,18 @@
     if (kafkaRestId != null) {
       withEnv("KAFKA_REST_ID", kafkaRestId);
     }
+    if (enableAuthentication) {
+      withEnv("KAFKA_REST_AUTHENTICATION_METHOD", "BASIC");
+      withEnv("KAFKA_REST_AUTHENTICATION_REALM", "KafkaRest");
+      withEnv("KAFKA_REST_AUTHENTICATION_ROLES", "GerritRole");
+      withEnv(
+          "KAFKAREST_OPTS",
+          "-Djava.security.auth.login.config=/etc/kafka-rest/rest-jaas.properties");
+      withClasspathResourceMapping(
+          "rest-jaas.properties", "/etc/kafka-rest/rest-jaas.properties", BindMode.READ_ONLY);
+      withClasspathResourceMapping(
+          "password.properties", "/etc/kafka-rest/password.properties", BindMode.READ_ONLY);
+    }
     withCreateContainerCmdModifier(cmd -> cmd.withHostName(kafkaRestHostname));
   }
 
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 37f15da..1991467 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
@@ -17,6 +17,7 @@
 import static com.google.common.truth.Truth.assertThat;
 import static org.mockito.Mockito.mock;
 
+import com.google.common.base.Strings;
 import com.google.gerrit.metrics.MetricMaker;
 import com.google.gerrit.server.events.Event;
 import com.google.gerrit.server.events.EventGson;
@@ -65,6 +66,8 @@
   static KafkaRestContainer kafkaRest;
   static KafkaRestContainer kafkaRestWithId;
   static GenericContainer<?> nginx;
+  static String restApiUsername;
+  static String restApiPassword;
 
   static final int TEST_NUM_SUBSCRIBERS = 1;
   static final String TEST_GROUP_ID = KafkaBrokerApiTest.class.getName();
@@ -167,9 +170,9 @@
   public static void beforeClass() throws Exception {
     kafka = KafkaContainerProvider.get();
     kafka.start();
-    kafkaRestWithId = new KafkaRestContainer(kafka, KAFKA_REST_ID);
+    kafkaRestWithId = new KafkaRestContainer(kafka, KAFKA_REST_ID, isAuthenticationProvided());
     kafkaRestWithId.start();
-    kafkaRest = new KafkaRestContainer(kafka);
+    kafkaRest = new KafkaRestContainer(kafka, isAuthenticationProvided());
     kafkaRest.start();
 
     System.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers());
@@ -194,6 +197,10 @@
     }
   }
 
+  private static boolean isAuthenticationProvided() {
+    return !Strings.isNullOrEmpty(restApiUsername) && !Strings.isNullOrEmpty(restApiPassword);
+  }
+
   protected TestModule newTestModule(KafkaProperties kafkaProperties) {
     return new TestModule(kafkaProperties);
   }
@@ -221,7 +228,9 @@
 
   @Test
   public void shouldSendSyncAndReceiveToTopic() {
-    connectToKafka(new KafkaProperties(false, clientType, getKafkaRestApiUriString()));
+    connectToKafka(
+        new KafkaProperties(
+            false, clientType, getKafkaRestApiUriString(), restApiUsername, restApiPassword));
     KafkaBrokerApi kafkaBrokerApi = injector.getInstance(KafkaBrokerApi.class);
     String testTopic = "test_topic_sync";
     TestConsumer testConsumer = new TestConsumer(1);
@@ -240,7 +249,9 @@
 
   @Test
   public void shouldSendAsyncAndReceiveToTopic() {
-    connectToKafka(new KafkaProperties(true, clientType, getKafkaRestApiUriString()));
+    connectToKafka(
+        new KafkaProperties(
+            true, clientType, getKafkaRestApiUriString(), restApiUsername, restApiPassword));
     KafkaBrokerApi kafkaBrokerApi = injector.getInstance(KafkaBrokerApi.class);
     String testTopic = "test_topic_async";
     TestConsumer testConsumer = new TestConsumer(1);
@@ -259,7 +270,9 @@
 
   @Test
   public void shouldSendToTopicAndResetOffset() {
-    connectToKafka(new KafkaProperties(false, clientType, getKafkaRestApiUriString()));
+    connectToKafka(
+        new KafkaProperties(
+            false, clientType, getKafkaRestApiUriString(), restApiUsername, restApiPassword));
     KafkaBrokerApi kafkaBrokerApi = injector.getInstance(KafkaBrokerApi.class);
     String testTopic = "test_topic_reset";
     Event testEventMessage = new ProjectCreatedEvent();
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
index c0c2bc7..5be61ab 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerRestApiTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerRestApiTest.java
@@ -56,7 +56,9 @@
                 TEST_GROUP_ID,
                 TEST_NUM_SUBSCRIBERS,
                 ClientType.REST,
-                getKafkaRestApiUriString());
+                getKafkaRestApiUriString(),
+                restApiUsername,
+                restApiPassword);
         bind(KafkaSubscriberProperties.class).toInstance(kafkaSubscriberProperties);
 
         bind(HttpHostProxy.class).toInstance(new HttpHostProxy(null, null, null));
diff --git a/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerRestApiTestBase.java b/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerRestApiTestBase.java
index 91b6286..cee073f 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerRestApiTestBase.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerRestApiTestBase.java
@@ -35,9 +35,11 @@
                 + "  listen  [::]:80  default_server;\\n"
                 + "  location     /%s/ {\\n"
                 + "    proxy_pass http://%s:%d/; \\n"
+                + "	   proxy_set_header Authorization \"Basic Z2Vycml0OnNlY3JldA==\";\\n"
                 + "  }\\n"
                 + "  location     / {\\n"
                 + "    proxy_pass http://%s:%d; \\n"
+                + "	   proxy_set_header Authorization \"Basic Z2Vycml0OnNlY3JldA==\";\\n"
                 + "  }\\n"
                 + "}",
             KAFKA_REST_ID,
diff --git a/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerRestApiWithAuthenticationTest.java b/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerRestApiWithAuthenticationTest.java
new file mode 100644
index 0000000..abd054f
--- /dev/null
+++ b/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerRestApiWithAuthenticationTest.java
@@ -0,0 +1,30 @@
+// Copyright (C) 2022 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 org.junit.BeforeClass;
+import org.junit.runner.RunWith;
+import org.mockito.junit.MockitoJUnitRunner;
+
+@RunWith(MockitoJUnitRunner.class)
+public class KafkaBrokerRestApiWithAuthenticationTest extends KafkaBrokerRestApiTest {
+
+  @BeforeClass
+  public static void beforeClass() throws Exception {
+    restApiUsername = "gerrit";
+    restApiPassword = "secret";
+    KafkaBrokerRestApiTest.beforeClass();
+  }
+}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerRestApiWithIdPrefixTest.java b/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerRestApiWithIdPrefixTest.java
index 9125989..6ee06e7 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerRestApiWithIdPrefixTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerRestApiWithIdPrefixTest.java
@@ -56,7 +56,9 @@
                 TEST_GROUP_ID,
                 TEST_NUM_SUBSCRIBERS,
                 ClientType.REST,
-                getApiUriString());
+                getApiUriString(),
+                null,
+                null);
         bind(KafkaSubscriberProperties.class).toInstance(kafkaSubscriberProperties);
 
         bind(HttpHostProxy.class).toInstance(new HttpHostProxy(null, null, null));
diff --git a/src/test/resources/password.properties b/src/test/resources/password.properties
new file mode 100644
index 0000000..bcff92d
--- /dev/null
+++ b/src/test/resources/password.properties
@@ -0,0 +1 @@
+gerrit: secret,GerritRole
\ No newline at end of file
diff --git a/src/test/resources/rest-jaas.properties b/src/test/resources/rest-jaas.properties
new file mode 100644
index 0000000..719a030
--- /dev/null
+++ b/src/test/resources/rest-jaas.properties
@@ -0,0 +1,5 @@
+KafkaRest {
+    org.eclipse.jetty.jaas.spi.PropertyFileLoginModule required
+    debug="true"
+    file="/etc/kafka-rest/password.properties";
+};
\ No newline at end of file