fix the historical on consuming

kafka_meta.go

@@ -26,26 +26,7 @@ func listenToMetadata(client *sarama.Client) {
 			if meta.Cancel != nil {
 				handleCancelRequest(meta.Cancel)
 			}
-			if meta.Restart != nil && !meta.Historical {
-				handleRestartRequest(meta.Restart)
-			}
-		}
-		msg.Processed <- "success"
-	}
-	log.Panic("listenToMetadada: Lost connection to Kafka")
-}
-
-//hack because we cant have history on the restart queue! another queue? another client?
-func listenToRestart(client *sarama.Client) {
-	topic := bn.ModuleToTopic(ModuleName, true)
-	ch := make(chan *kafka.ConsumerData)
-	go kafka.Consume(client, topic, ch)
-	for msg := range ch {
-		var meta pb.Meta
-		if err := proto.Unmarshal(msg.Value, &meta); err != nil {
-			log.WithError(err).Warn("Unknown meta data received")
-		} else {
-			if meta.Restart != nil {
+			if meta.Restart != nil && !msg.Historical {
 				handleRestartRequest(meta.Restart)
 			}
 		}

main.go

@@ -57,9 +57,6 @@ func run(address string, kafkaBrokers []string) {
 		client := kafka.NewClient(kafkaBrokers, clientID, debug)
 		defer (*client).Close()
 
-		clientRestart := kafka.NewClient(kafkaBrokers, clientID+".restart", debug)
-		defer (*clientRestart).Close()
-
 		clientgroup := kafka.NewClientGroup(kafkaBrokers, clientID, debug)
 		defer (*clientgroup).Close()
 
@@ -73,7 +70,6 @@ func run(address string, kafkaBrokers []string) {
 		go registerModule(producer, ModuleName)
 		go startWorker(clientgroup, producer, []string{module.KafkaTopicName("IO"), module.KafkaTopicName("Ping")})
 		go listenToMetadata(client)
-		go listenToRestart(clientRestart)
 	}
 
 	log.WithFields(log.Fields{"version": Version, "interfaces": interfaces, "debug": debug}).Warn("Start module")