func Start_topic_reader(topic string, type_id string, control_ch chan dataTypes.ReaderControl, data_ch chan *dataTypes.KafkaPayload, gid string, cid string) {
log.Info("Topic reader starting, topic: ", topic, " for type: ", type_id)
func Start_topic_reader(topic string, type_id string, control_ch chan dataTypes.ReaderControl, data_ch chan *dataTypes.KafkaPayload, gid string, cid string) {
log.Info("Topic reader starting, topic: ", topic, " for type: ", type_id)