Manage Kafka clientType when starting session

Perform different checks when starting the Kafka client
session for different clientType values.

Change-Id: If7e498837dc0bf2bf8acc31f3e5276fe223646e2

src/main/java/com/googlesource/gerrit/plugins/kafka/session/KafkaSession.java

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 f7105f9..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,6 +18,7 @@
 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.Producer;
 import org.apache.kafka.clients.producer.ProducerRecord;
@@ -56,18 +57,37 @@
       return;
     }
 
-    String bootstrapServers = properties.getProperty("bootstrap.servers");
-    if (bootstrapServers == null) {
-      LOGGER.warn("No Kafka bootstrap.servers property defined: session not started.");
-      return;
+    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;
     }
 
-    LOGGER.info("Connect to {}...", bootstrapServers);
-    /* 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();
     producer = producerProvider.get();
     LOGGER.info("Connection established.");
   }

src/test/java/com/googlesource/gerrit/plugins/kafka/publish/KafkaSessionTest.java

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 ccf60ce..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
@@ -22,6 +22,7 @@
 
 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;
@@ -54,6 +55,7 @@
     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);