fix serializeKey function

smueller18 · Sep 25, 2017 · 830efc0 · 830efc0
1 parent 0f12060
commit 830efc0
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 7 deletions.
diff --git 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/...n/java/com/github/smueller18/flink/serialization/AvroBuilderKeyedSerializationSchema.java b/...n/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());
     }