➕ Upgrading kafka-client-go, using new Consume func which will block …

…while consuming messages:
Financial-Times · Nov 22, 2018 · 6a05d95 · 6a05d95
1 parent db23258
commit 6a05d95
Show file tree

Hide file tree

Showing 4 changed files with 12 additions and 5 deletions.
diff --git a/Gopkg.lock b/Gopkg.lock
diff --git a/Gopkg.toml b/Gopkg.toml
@@ -35,7 +35,7 @@
 
 [[constraint]]
   name = "github.com/Financial-Times/kafka-client-go"
-  version = "0.8.3"
+  version = "0.9.0-blocking-consume-rc1"
 
 [[constraint]]
   name = "github.com/Financial-Times/service-status-go"

diff --git a/healthcheck_test.go b/healthcheck_test.go
@@ -26,6 +26,10 @@ func (mc mockKafkaConnection) StartListening(messageHandler func(message kafka.F
 	return
 }
 
+func (mc mockKafkaConnection) Consume(messageHandler func(message kafka.FTMessage) error) {
+	return
+}
+
 func (mc mockKafkaConnection) SendMessage(message kafka.FTMessage) error {
 	return nil
 }

diff --git a/kafka.go b/kafka.go
@@ -13,7 +13,10 @@ import (
 )
 
 func startKafkaConsumer(messageConsumer kafka.Consumer) {
-	messageConsumer.StartListening(handleMessage)
+	for {
+		messageConsumer.Consume(handleMessage)
+		logger.Warnf(nil, "Consumer stopped processing messages, restarting the consumer.")
+	}
 }
 
 func handleMessage(msg kafka.FTMessage) error {