From 830efc0ed185872fa378b070f87e47cce86706fc Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Stephan=20M=C3=BCller?= <mail@stephanmueller.eu>
Date: Mon, 25 Sep 2017 16:09:48 +0200
Subject: [PATCH] fix serializeKey function

---
 pom.xml                                              |  2 +-
 .../AvroBuilderKeyedSerializationSchema.java         | 12 ++++++------
 2 files changed, 7 insertions(+), 7 deletions(-)

diff --git a/pom.xml b/pom.xml
index a96d1d3..3298415 100644
--- a/pom.xml
+++ b/pom.xml
@@ -6,7 +6,7 @@
 
     <groupId>com.github.smueller18</groupId>
     <artifactId>flink-serialization</artifactId>
-    <version>0.10.0</version>
+    <version>0.10.1</version>
     <packaging>jar</packaging>
 
     <name>Flink-Serialization</name>
diff --git a/src/main/java/com/github/smueller18/flink/serialization/AvroBuilderKeyedSerializationSchema.java b/src/main/java/com/github/smueller18/flink/serialization/AvroBuilderKeyedSerializationSchema.java
index a7f353b..d3dc8ee 100644
--- a/src/main/java/com/github/smueller18/flink/serialization/AvroBuilderKeyedSerializationSchema.java
+++ b/src/main/java/com/github/smueller18/flink/serialization/AvroBuilderKeyedSerializationSchema.java
@@ -55,9 +55,9 @@ public AvroBuilderKeyedSerializationSchema(Map<String, ?> configs) {
     @Override
     public byte[] serializeKey(T avroBuilderObject) {
 
-        if (this.valueSerializer == null) {
-            this.valueSerializer = new KafkaAvroSerializer();
-            this.valueSerializer.configure(this.configs, true);
+        if (this.keySerializer == null) {
+            this.keySerializer = new KafkaAvroSerializer();
+            this.keySerializer.configure(this.configs, true);
         }
 
         return this.keySerializer.serialize(AvroBuilder.getTopicName(avroBuilderObject.getClass()), avroBuilderObject.getKeyRecord());
@@ -66,9 +66,9 @@ public byte[] serializeKey(T avroBuilderObject) {
     @Override
     public byte[] serializeValue(T avroBuilderObject) {
 
-        if (this.keySerializer == null) {
-            this.keySerializer = new KafkaAvroSerializer();
-            this.keySerializer.configure(this.configs, false);
+        if (this.valueSerializer == null) {
+            this.valueSerializer = new KafkaAvroSerializer();
+            this.valueSerializer.configure(this.configs, false);
         }
         return this.valueSerializer.serialize(AvroBuilder.getTopicName(avroBuilderObject.getClass()), avroBuilderObject.getValueRecord());
     }