Added configurable topic options for Kafka driver

merlimat · merlimat · commit a5de8562a030 · 2017-12-13T11:26:48.000-08:00
diff --git a/driver-kafka/kafka-exactly-once.yaml b/driver-kafka/kafka-exactly-once.yaml
@@ -25,6 +25,9 @@ driverClass: io.openmessaging.benchmark.driver.kafka.KafkaBenchmarkDriver
 
 replicationFactor: 3
 
+topicConfig: |
+	min.insync.replicas=2
+
 commonConfig: |
     bootstrap.servers=localhost:9092
 
diff --git a/driver-kafka/kafka-sync.yaml b/driver-kafka/kafka-sync.yaml
@@ -0,0 +1,45 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you 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.
+#
+
+# Kafka driver profile that is configuring to sync all
+# published messages to disk to achieve durability
+
+name : Kafka
+driverClass: io.openmessaging.benchmark.driver.kafka.KafkaBenchmarkDriver
+
+# Kafka client specific configurations
+
+replicationFactor: 3
+
+topicConfig: |
+    min.insync.replicas=2
+    flush.messages=1
+    flush.ms=0
+
+commonConfig: |
+    bootstrap.servers=localhost:9092
+
+producerConfig: |
+    acks=all
+    linger.ms=1
+    batch.size=131072
+
+consumerConfig: |
+    auto.offset.reset=earliest
+    enable.auto.commit=false
diff --git a/driver-kafka/kafka.yaml b/driver-kafka/kafka.yaml
@@ -25,6 +25,9 @@ driverClass: io.openmessaging.benchmark.driver.kafka.KafkaBenchmarkDriver
 
 replicationFactor: 3
 
+topicConfig: |
+	min.insync.replicas=2
+
 commonConfig: |
     bootstrap.servers=localhost:9092
 
diff --git a/driver-kafka/src/main/java/io/openmessaging/benchmark/driver/kafka/Config.java b/driver-kafka/src/main/java/io/openmessaging/benchmark/driver/kafka/Config.java
@@ -21,6 +21,8 @@
 public class Config {
     public short replicationFactor;
 
+    public String topicConfig;
+
     public String commonConfig;
 
     public String producerConfig;
diff --git a/driver-kafka/src/main/java/io/openmessaging/benchmark/driver/kafka/KafkaBenchmarkDriver.java b/driver-kafka/src/main/java/io/openmessaging/benchmark/driver/kafka/KafkaBenchmarkDriver.java
@@ -24,7 +24,9 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.Properties;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
@@ -56,6 +58,7 @@ public class KafkaBenchmarkDriver implements BenchmarkDriver {
     private KafkaProducer<String, byte[]> producer;
     private List<BenchmarkConsumer> consumers = Collections.synchronizedList(new ArrayList<>());
 
+    private Properties topicProperties;
     private Properties producerProperties;
     private Properties consumerProperties;
 
@@ -80,6 +83,9 @@ public void initialize(File configurationFile) throws IOException {
         consumerProperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
         consumerProperties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
 
+        topicProperties = new Properties();
+        topicProperties.load(new StringReader(config.topicConfig));
+
         admin = AdminClient.create(commonProperties);
 
         producer = new KafkaProducer<>(producerProperties);
@@ -90,12 +96,14 @@ public String getTopicNamePrefix() {
         return "test-topic";
     }
 
+    @SuppressWarnings({ "rawtypes", "unchecked" })
     @Override
     public CompletableFuture<Void> createTopic(String topic, int partitions) {
         return CompletableFuture.runAsync(() -> {
             try {
-                admin.createTopics(Arrays.asList(new NewTopic(topic, partitions, config.replicationFactor))).all()
-                        .get();
+                NewTopic newTopic = new NewTopic(topic, partitions, config.replicationFactor);
+                newTopic.configs(new HashMap<>((Map) topicProperties));
+                admin.createTopics(Arrays.asList(newTopic)).all().get();
             } catch (InterruptedException | ExecutionException e) {
                 throw new RuntimeException(e);
             }
