kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ewe...@apache.org
Subject kafka git commit: KAFKA-3134: Fix missing value.deserializer error during KafkaConsumer initialization
Date Sun, 24 Jan 2016 09:55:52 GMT
Repository: kafka
Updated Branches:
  refs/heads/trunk c8b60b634 -> 5e8a08483


KAFKA-3134: Fix missing value.deserializer error during KafkaConsumer initialization

… initialization

Author: Yifan Ying <yying@fitbit.com>

Reviewers: Ismael Juma <ismael@juma.me.uk>, Grant Henke <granthenke@gmail.com>,
Ewen Cheslack-Postava <ewen@confluent.io>

Closes #803 from happymap/KAFKA-3134


Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/5e8a0848
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/5e8a0848
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/5e8a0848

Branch: refs/heads/trunk
Commit: 5e8a084834ad35506ee74e1da15a3964642a512e
Parents: c8b60b6
Author: Yifan Ying <yying@fitbit.com>
Authored: Sun Jan 24 01:55:30 2016 -0800
Committer: Ewen Cheslack-Postava <me@ewencp.org>
Committed: Sun Jan 24 01:55:30 2016 -0800

----------------------------------------------------------------------
 .../kafka/clients/consumer/ConsumerConfig.java  |  4 +-
 .../clients/consumer/ConsumerConfigTest.java    | 92 ++++++++++++++++++++
 2 files changed, 94 insertions(+), 2 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/5e8a0848/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java
index 4131352..9355b08 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java
@@ -315,7 +315,7 @@ public class ConsumerConfig extends AbstractConfig {
         newConfigs.putAll(configs);
         if (keyDeserializer != null)
             newConfigs.put(KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializer.getClass());
-        if (keyDeserializer != null)
+        if (valueDeserializer != null)
             newConfigs.put(VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer.getClass());
         return newConfigs;
     }
@@ -327,7 +327,7 @@ public class ConsumerConfig extends AbstractConfig {
         newProperties.putAll(properties);
         if (keyDeserializer != null)
             newProperties.put(KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializer.getClass().getName());
-        if (keyDeserializer != null)
+        if (valueDeserializer != null)
             newProperties.put(VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer.getClass().getName());
         return newProperties;
     }

http://git-wip-us.apache.org/repos/asf/kafka/blob/5e8a0848/clients/src/test/java/org/apache/kafka/clients/consumer/ConsumerConfigTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/ConsumerConfigTest.java
b/clients/src/test/java/org/apache/kafka/clients/consumer/ConsumerConfigTest.java
new file mode 100644
index 0000000..5a7699d
--- /dev/null
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/ConsumerConfigTest.java
@@ -0,0 +1,92 @@
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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 org.apache.kafka.clients.consumer;
+
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+import org.apache.kafka.common.serialization.Deserializer;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.junit.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.junit.Assert.assertEquals;
+
+public class ConsumerConfigTest {
+
+    private final Deserializer keyDeserializer = new ByteArrayDeserializer();
+    private final Deserializer valueDeserializer = new StringDeserializer();
+    private final String keyDeserializerClassName = keyDeserializer.getClass().getName();
+    private final String valueDeserializerClassName = valueDeserializer.getClass().getName();
+    private final Object keyDeserializerClass = keyDeserializer.getClass();
+    private final Object valueDeserializerClass = valueDeserializer.getClass();
+
+    @Test
+    public void testDeserializerToPropertyConfig() {
+        Properties properties = new Properties();
+        properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializerClassName);
+        properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializerClassName);
+        Properties newProperties = ConsumerConfig.addDeserializerToConfig(properties, null,
null);
+        assertEquals(newProperties.get(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG), keyDeserializerClassName);
+        assertEquals(newProperties.get(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG), valueDeserializerClassName);
+
+        properties.clear();
+        properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializerClassName);
+        newProperties = ConsumerConfig.addDeserializerToConfig(properties, keyDeserializer,
null);
+        assertEquals(newProperties.get(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG), keyDeserializerClassName);
+        assertEquals(newProperties.get(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG), valueDeserializerClassName);
+
+        properties.clear();
+        properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializerClassName);
+        newProperties = ConsumerConfig.addDeserializerToConfig(properties, null, valueDeserializer);
+        assertEquals(newProperties.get(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG), keyDeserializerClassName);
+        assertEquals(newProperties.get(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG), valueDeserializerClassName);
+
+        properties.clear();
+        newProperties = ConsumerConfig.addDeserializerToConfig(properties, keyDeserializer,
valueDeserializer);
+        assertEquals(newProperties.get(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG), keyDeserializerClassName);
+        assertEquals(newProperties.get(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG), valueDeserializerClassName);
+    }
+
+    @Test
+    public void testDeserializerToMapConfig() {
+        Map<String, Object> configs = new HashMap<>();
+        configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializerClass);
+        configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializerClass);
+        Map<String, Object> newConfigs = ConsumerConfig.addDeserializerToConfig(configs,
null, null);
+        assertEquals(newConfigs.get(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG), keyDeserializerClass);
+        assertEquals(newConfigs.get(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG), valueDeserializerClass);
+
+        configs.clear();
+        configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializerClass);
+        newConfigs = ConsumerConfig.addDeserializerToConfig(configs, keyDeserializer, null);
+        assertEquals(newConfigs.get(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG), keyDeserializerClass);
+        assertEquals(newConfigs.get(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG), valueDeserializerClass);
+
+        configs.clear();
+        configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializerClass);
+        newConfigs = ConsumerConfig.addDeserializerToConfig(configs, null, valueDeserializer);
+        assertEquals(newConfigs.get(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG), keyDeserializerClass);
+        assertEquals(newConfigs.get(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG), valueDeserializerClass);
+
+        configs.clear();
+        newConfigs = ConsumerConfig.addDeserializerToConfig(configs, keyDeserializer, valueDeserializer);
+        assertEquals(newConfigs.get(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG), keyDeserializerClass);
+        assertEquals(newConfigs.get(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG), valueDeserializerClass);
+    }
+}


Mime
View raw message