Merge "Kafka REST Client: avoid clashes between clients" into stable-3.3

src/main/java/com/googlesource/gerrit/plugins/kafka/rest/KafkaRestClient.java

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 018f072..ccc801b 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
@@ -160,7 +160,7 @@
     post.setConfig(createRequestConfig());
     post.setEntity(
         new StringEntity(
-            "{\"format\": \"json\",\"auto.offset.reset\": \"earliest\"}",
+            "{\"format\": \"json\",\"auto.offset.reset\": \"earliest\", \"auto.commit.enable\":\"true\", \"consumer.request.timeout.ms\": \"1000\"}",
             ContentType.create(KAFKA_V2, UTF_8)));
     return post;
   }

src/main/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventRestSubscriber.java

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
index 9fdbbaa..1b638a3 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventRestSubscriber.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventRestSubscriber.java
@@ -260,8 +260,8 @@
       return restClient.execute(delete);
     }
 
-    private ListenableFuture<URI> createConsumer(String name) {
-      HttpPost post = restClient.createPostToConsumer(name);
+    private ListenableFuture<URI> createConsumer(String consumerGroup) {
+      HttpPost post = restClient.createPostToConsumer(consumerGroup + "-" + topic);
       return restClient.mapAsync(restClient.execute(post, HttpStatus.SC_OK), this::getConsumerUri);
     }