title: How to process and aggregate Kafka Streams with custom Objects?
body: <p>So basically I have Account class. I have data. I want to send those objects into my topic with producer. That is okay for now. Later on, I want to do aggregation with Kafka Streams but I can not because some Serde properties is wrong in my configuration, I think :/. I dont know where the error is. My producer works fine, but I can't aggregate. Anyone help me to look my kafka streams code please?
   My Account class:</p>
  
  <pre><code>public class Account {
  
      private long fromId;
      private long amount;
      private long toId;
      private ZonedDateTime time;
  }
  </code></pre>
  
  <p>There 2 classes Serializer and Deserializer for my Account class. Serializer:</p>
  
  <pre><code>public class AccountSerializer implements Serializer {
  
      private static final Charset CHARSET = Charset.forName("UTF-8");
      static private Gson gson = new Gson();
  
  
      @Override
      public void configure(Map map, boolean b) {
  
      }
  
      @Override
      public byte[] serialize(String s, Object o) {
          String line = gson.toJson(o);
          // Return the bytes from the String 'line'
          return line.getBytes(CHARSET);
      }
  
  
  
      @Override
      public void close() {
  
      }
  }
  </code></pre>
  
  <p>Deserializer:</p>
  
  <pre><code>public class AccountDeserializer implements Deserializer {
      private static final Charset CHARSET = Charset.forName("UTF-8");
      static private Gson gson;
  
      static {
          gson = new Gson();
      }
  
      @Override
      public void configure(Map map, boolean b) {
  
      }
  
      @Override
      public Object deserialize(String s, byte[] bytes) {
          try {
              // Transform the bytes to String
              String person = new String(bytes, CHARSET);
              // Return the Person object created from the String 'person'
              return gson.fromJson(person, Account.class);
          } catch (Exception e) {
              throw new IllegalArgumentException("Error reading bytes! Yanlış", e);
          }
      }
  
      @Override
      public void close() {
  
      }
  }
  </code></pre>
  
  <p>My AccountSerde class for kafka streams:</p>
  
  <pre><code>public class AccountSerde implements Serde&lt;Object&gt; {
  
      private AccountSerializer accountSerializer;
      private AccountDeserializer accountDeserializer;
  
      @Override
      public void configure(Map&lt;String, ?&gt; map, boolean b) {
  
  
      }
  
      @Override
      public void close() {
          accountSerializer.close();
          accountDeserializer.close();
  
      }
  
      @Override
      public Serializer&lt;Object&gt; serializer() {
          return accountSerializer;
      }
  
      @Override
      public Deserializer&lt;Object&gt; deserializer() {
          return accountDeserializer;
      }
  }
  </code></pre>
  
  <p>And my Kafka Producer:</p>
  
  <pre><code> public static void main(String[] args) {
  
          DataAccess dataAccess = new DataAccess();
          List&lt;Account&gt; accountList = dataAccess.read();
  
          final Logger logger = LoggerFactory.getLogger(Producer.class);
          Properties properties = new Properties();
  
          properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"127.0.0.1:9092");
          properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,LongSerializer.class.getName());
          properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,AccountSerializer.class.getName());
  
  
          KafkaProducer&lt;Long,Account&gt; producer = new KafkaProducer&lt;&gt;(properties);
  
  
  
          for (Account account : accountList) {
  
              ProducerRecord&lt;Long,Account&gt; record = new ProducerRecord&lt;Long, Account&gt;("bank_account",account.getFromId(),account);
  
  
              producer.send(record, new Callback() {
                  public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                      if (e == null) {
  
  
              logger.info("Record sent successfully. \n "+ "Topic : "+recordMetadata.topic() +"\n"+
                              "Partition : " + recordMetadata.partition() + "\n"+
                              "Offset : " +recordMetadata.offset() +"\n"+
                              "Timestamp: " +recordMetadata.timestamp() +"\n");
                      try {
                          Thread.sleep(1000);
                      } catch (InterruptedException e1) {
                          e1.printStackTrace();
                      }
  
                  }
                  else{
                      logger.info("Error sending producer");
                  }
              }
          });
      }
  
  
      producer.flush();
      producer.close();
  }
  </code></pre>
  
  <p>And here is class where I want to try aggregation, my Kafka Stream class.</p>
  
  <pre><code>  public static void main(String[] args) {
          System.out.println();
  
          Properties properties = new Properties();
          properties.setProperty(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"127.0.01:9092");
          properties.setProperty(StreamsConfig.APPLICATION_ID_CONFIG,"demo-kafka-streams");
          properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,AccountDeserializer.class.getName());
          properties.setProperty(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.LongSerde);
          properties.setProperty(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, AccountSerde.class.getName());
          //create a topology
  
          StreamsBuilder streamsBuilder = new StreamsBuilder();
  
          KStream&lt;Long, Account&gt; inputTopic = streamsBuilder.stream("bank_account");
  
          KTable&lt;Long, Long&gt; aggregate = inputTopic.groupByKey().aggregate(
                  () -&gt; 0L,
                  (key, current, oldBalance) -&gt; current.getAmount() + oldBalance);
  
          aggregate.toStream().to("son");
  
          KafkaStreams streams = new KafkaStreams(streamsBuilder.build(),properties);
  
          streams.start();
  
          System.out.println(streams.toString());
  
          Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
  
  
      }
  </code></pre>
  
  <p>I tried my producer is working fine and sends objects. However because of error I can't try whether my aggregation code is working or not. It gives me </p>
  
  <pre><code>[demo-kafka-streams-9e3b0ab8-c021-4707-bf85-174e1356ea89-StreamThread-1] ERROR org.apache.kafka.streams.errors.LogAndFailExceptionHandler - Exception caught during Deserialization, taskId: 0_0, topic: bank_account, partition: 0, offset: 0
  java.lang.NullPointerException
      at org.apache.kafka.streams.processor.internals.SourceNode.deserializeValue(SourceNode.java:63)
      at org.apache.kafka.streams.processor.internals.RecordDeserializer.deserialize(RecordDeserializer.java:66)
      at org.apache.kafka.streams.processor.internals.RecordQueue.addRawRecords(RecordQueue.java:97)
      at org.apache.kafka.streams.processor.internals.PartitionGroup.addRawRecords(PartitionGroup.java:117)
      at org.apache.kafka.streams.processor.internals.StreamTask.addRecords(StreamTask.java:638)
      at org.apache.kafka.streams.processor.internals.StreamThread.addRecordsToTasks(StreamThread.java:936)
      at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:831)
      at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:767)
      at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:736)
  [demo-kafka-streams-9e3b0ab8-c021-4707-bf85-174e1356ea89-StreamThread-1] INFO org.apache.kafka.streams.processor.internals.StreamThread - stream-thread [demo-kafka-streams-9e3b0ab8-c021-4707-bf85-174e1356ea89-StreamThread-1] State transition from RUNNING to PENDING_SHUTDOWN
  [demo-kafka-streams-9e3b0ab8-c021-4707-bf85-174e1356ea89-StreamThread-1] INFO org.apache.kafka.streams.processor.internals.StreamThread - stream-thread [demo-kafka-streams-9e3b0ab8-c021-4707-bf85-174e1356ea89-StreamThread-1] Shutting down
  [demo-kafka-streams-9e3b0ab8-c021-4707-bf85-174e1356ea89-StreamThread-1] INFO org.apache.kafka.clients.producer.KafkaProducer - [Producer clientId=demo-kafka-streams-9e3b0ab8-c021-4707-bf85-174e1356ea89-StreamThread-1-producer] Closing the Kafka producer with timeoutMillis = 9223372036854775807 ms.
  [demo-kafka-streams-9e3b0ab8-c021-4707-bf85-174e1356ea89-StreamThread-1] INFO org.apache.kafka.streams.processor.internals.StreamThread - stream-thread [demo-kafka-streams-9e3b0ab8-c021-4707-bf85-174e1356ea89-StreamThread-1] State transition from PENDING_SHUTDOWN to DEAD
  [demo-kafka-streams-9e3b0ab8-c021-4707-bf85-174e1356ea89-StreamThread-1] INFO org.apache.kafka.streams.KafkaStreams - stream-client [demo-kafka-streams-9e3b0ab8-c021-4707-bf85-174e1356ea89] State transition from RUNNING to ERROR
  [demo-kafka-streams-9e3b0ab8-c021-4707-bf85-174e1356ea89-StreamThread-1] WARN org.apache.kafka.streams.KafkaStreams - stream-client [demo-kafka-streams-9e3b0ab8-c021-4707-bf85-174e1356ea89] All stream threads have died. The instance will be in error state and should be closed.
  [demo-kafka-streams-9e3b0ab8-c021-4707-bf85-174e1356ea89-StreamThread-1] INFO org.apache.kafka.streams.processor.internals.StreamThread - stream-thread [demo-kafka-streams-9e3b0ab8-c021-4707-bf85-174e1356ea89-StreamThread-1] Shutdown complete
  Exception in thread "demo-kafka-streams-9e3b0ab8-c021-4707-bf85-174e1356ea89-StreamThread-1" org.apache.kafka.streams.errors.StreamsException: Deserialization exception handler is set to fail upon a deserialization error. If you would rather have the streaming pipeline continue after a deserialization error, please set the default.deserialization.exception.handler appropriately.
      at org.apache.kafka.streams.processor.internals.RecordDeserializer.deserialize(RecordDeserializer.java:80)
      at org.apache.kafka.streams.processor.internals.RecordQueue.addRawRecords(RecordQueue.java:97)
      at org.apache.kafka.streams.processor.internals.PartitionGroup.addRawRecords(PartitionGroup.java:117)
      at org.apache.kafka.streams.processor.internals.StreamTask.addRecords(StreamTask.java:638)
      at org.apache.kafka.streams.processor.internals.StreamThread.addRecordsToTasks(StreamThread.java:936)
      at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:831)
      at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:767)
      at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:736)
  Caused by: java.lang.NullPointerException
      at org.apache.kafka.streams.processor.internals.SourceNode.deserializeValue(SourceNode.java:63)
      at org.apache.kafka.streams.processor.internals.RecordDeserializer.deserialize(RecordDeserializer.java:66)
      ... 7 more
  </code></pre>