updated to also write to kafka brokers