Fixed kafka key serializer class

merlimat · merlimat · commit 93ac64cd9c80 · 2017-12-02T09:18:23.000-08:00
diff --git a/driver-kafka/src/main/java/io/openmessaging/benchmark/driver/kafka/KafkaBenchmarkConsumer.java b/driver-kafka/src/main/java/io/openmessaging/benchmark/driver/kafka/KafkaBenchmarkConsumer.java
@@ -35,22 +35,22 @@
 
 public class KafkaBenchmarkConsumer implements BenchmarkConsumer {
 
-    private final KafkaConsumer<byte[], byte[]> consumer;
+    private final KafkaConsumer<String, byte[]> consumer;
 
     private final ExecutorService executor;
     private final Future<?> consumerTask;
     private volatile boolean closing = false;
 
-    public KafkaBenchmarkConsumer(KafkaConsumer<byte[], byte[]> consumer, ConsumerCallback callback) {
+    public KafkaBenchmarkConsumer(KafkaConsumer<String, byte[]> consumer, ConsumerCallback callback) {
         this.consumer = consumer;
         this.executor = Executors.newSingleThreadExecutor();
 
         this.consumerTask = this.executor.submit(() -> {
             while (!closing) {
-                ConsumerRecords<byte[], byte[]> records = consumer.poll(100);
+                ConsumerRecords<String, byte[]> records = consumer.poll(100);
 
                 Map<TopicPartition, OffsetAndMetadata> offsetMap = new HashMap<>();
-                for (ConsumerRecord<byte[], byte[]> record : records) {
+                for (ConsumerRecord<String, byte[]> record : records) {
                     callback.messageReceived(record.value(), record.timestamp());
 
                     offsetMap.put(new TopicPartition(record.topic(), record.partition()),
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
@@ -37,6 +37,8 @@
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.common.serialization.ByteArrayDeserializer;
 import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
 
 import com.fasterxml.jackson.databind.DeserializationFeature;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -69,13 +71,13 @@ public void initialize(File configurationFile) throws IOException {
         producerProperties = new Properties();
         commonProperties.forEach((key, value) -> producerProperties.put(key, value));
         producerProperties.load(new StringReader(config.producerConfig));
-        producerProperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
+        producerProperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
         producerProperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
 
         consumerProperties = new Properties();
         commonProperties.forEach((key, value) -> consumerProperties.put(key, value));
         consumerProperties.load(new StringReader(config.consumerConfig));
-        consumerProperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
+        consumerProperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
         consumerProperties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
 
         admin = AdminClient.create(commonProperties);
@@ -111,7 +113,7 @@ public CompletableFuture<BenchmarkConsumer> createConsumer(String topic, String
         Properties properties = new Properties();
         consumerProperties.forEach((key, value) -> properties.put(key, value));
         properties.put(ConsumerConfig.GROUP_ID_CONFIG, subscriptionName);
-        KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<>(properties);
+        KafkaConsumer<String, byte[]> consumer = new KafkaConsumer<>(properties);
         try {
             consumer.subscribe(Arrays.asList(topic));
             return CompletableFuture.completedFuture(new KafkaBenchmarkConsumer(consumer, consumerCallback));
