Use KStream#merge for SourceStatisticsMonitor

[blootsvoets]

Instead of implementing source statistics as a monitor, we could implement it as a stream by using KStream#merge(KStream). This would allow using a stream StateStore to store the statistics in Kafka.