Moar refactor

streams/src/main/java/io/confluent/demo/StreamsDemo.java

@@ -25,8 +25,9 @@
 import java.util.Map;
 import java.util.Properties;
 
-import io.confluent.demo.util.CountAndSum;
 import io.confluent.demo.util.CountAndSumDeserializer;
+import io.confluent.demo.util.CountAndSumSerde;
+import io.confluent.demo.util.CountAndSum;
 import io.confluent.demo.util.CountAndSumSerializer;
 import io.confluent.kafka.serializers.AbstractKafkaAvroSerDeConfig;
 import io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde;
@@ -152,7 +153,7 @@ protected static KTable<Long, Double> getRatingAverageTable(KStream<Long, String
     KGroupedStream<Long, Double> ratingsById = ratings.mapValues(Rating::getRating).groupByKey();
 
     /*
-    as per Matthias' comment https://issues.apache.org/jira/browse/KAFKA-7595?focusedCommentId=16677173&page=com.atlassian.jira.plugin.system.issuetabpanels%3Acomment-tabpanel#comment-16677173
+    As per Matthias' comment https://issues.apache.org/jira/browse/KAFKA-7595?focusedCommentId=16677173&page=com.atlassian.jira.plugin.system.issuetabpanels%3Acomment-tabpanel#comment-16677173
     implementation
 
     KTable<Long, Long> ratingCounts = ratingsById.count();
@@ -168,26 +169,17 @@ protected static KTable<Long, Double> getRatingAverageTable(KStream<Long, String
           ++aggregate.count;
           aggregate.sum += value;
           return aggregate;
-        }, Materialized.with(new LongSerde(),
-                             new Serde<CountAndSum<Long, Double>>() {
-                               @Override
-                               public void configure(Map<String, ?> configs, boolean isKey) {
-                               }
-
-                               @Override
-                               public void close() {
-                               }
-
-                               @Override
-                               public Serializer<CountAndSum<Long, Double>> serializer() {
-                                 return new CountAndSumSerializer<>();
-                               }
-
-                               @Override
-                               public Deserializer<CountAndSum<Long, Double>> deserializer() {
-                                 return new CountAndSumDeserializer<>();
-                               }
-                             })
+        }, Materialized.with(new LongSerde(), new CountAndSumSerde<Long, Double>() {
+          @Override
+          public Serializer<CountAndSum<Long, Double>> serializer() {
+            return new CountAndSumSerializer<>();
+          }
+
+          @Override
+          public Deserializer<CountAndSum<Long, Double>> deserializer() {
+            return new CountAndSumDeserializer<>();
+          }
+        })
     );
 
     /*KTable<Long, Double> ratingAverage = ratingSums.join(ratingCounts,

streams/src/main/java/io/confluent/demo/util/CountAndSumSerde.java

@@ -0,0 +1,18 @@
+package io.confluent.demo.util;
+
+import org.apache.kafka.common.serialization.Serde;
+
+import java.util.Map;
+
+public abstract class CountAndSumSerde<T1, T2> implements Serde<CountAndSum<T1, T2>> {
+
+  @Override
+  public void configure(Map<String, ?> configs, boolean isKey) {
+
+  }
+
+  @Override
+  public void close() {
+
+  }
+}