each

...
if (pushNotificationResponse.isAccepted()) {
  logger.trace("Push notification for '{}' (payload={})", deviceToken, pushNotificationResponse.getPushNotification().getPayload());
 
  producer.send(jobID, "Success"); // sends to "push_messages" topic
} else {
  final String rejectReason = pushNotificationResponse.getRejectionReason();
  logger.trace("Push Message has been rejected with reason: {}", rejectReason);
 
  producer.send(jobID, "Rejected"); // sends "push_messages" topic
...
}
KStreamBuilder builder = new KStreamBuilder();
 
// read from the topic that contains all messages, for all jobs
final KStream<String, String> source = builder.stream("push_messages");
 
 
// some simple processing, and grouping by key, applying a predicate and send to three "analytic" topic:
 
final KTable<String, Long> successCountsPerJob = source.filter((key, value) -> value.equals("Success"))
  .groupByKey()
  .count("successMessagesPerJob");
successCountsPerJob.to(Serdes.String(), Serdes.Long(), "successMessagesPerJob");
 
final KTable<String, Long> failCountsPerJob = source.filter((key, value) -> value.equals("Rejected"))
  .groupByKey()
  .count("failedMessagesPerJob");
failCountsPerJob.to(Serdes.String(), Serdes.Long(), "failedMessagesPerJob");
 
source.groupByKey()
  count("totalMessagesPerJob")
    .to(Serdes.String(), Serdes.Long(), "totalMessagesPerJob");
 
2017-05-16 13:42:48,763 INFO  successMessagesPerJob: 2 - jobID: XXX
2017-05-16 13:42:48,764 INFO  totalMessagesPerJob: 3 - jobID: XXX
2017-05-16 13:42:48,764 INFO  failedMessagesPerJob: 1 - jobID: XXX