Update service to use v3 of kafka-client-go

app.go

@@ -35,11 +35,18 @@ func main() {
 		EnvVar: "NOTIFICATIONS_RESOURCE",
 	})
 
-	consumerAddrs := app.String(cli.StringOpt{
+	consumerAddress := app.String(cli.StringOpt{
 		Name:   "consumer_addr",
 		Value:  "",
 		Desc:   "Comma separated kafka hosts for message consuming.",
-		EnvVar: "KAFKA_ADDRS",
+		EnvVar: "KAFKA_ADDRESS",
+	})
+
+	consumerLagTolerance := app.Int(cli.IntOpt{
+		Name:   "consumer_lag_tolerance",
+		Value:  120,
+		Desc:   "Kafka lag tolerance",
+		EnvVar: "KAFKA_LAG_TOLERANCE",
 	})
 
 	consumerGroupID := app.String(cli.StringOpt{
@@ -163,7 +170,8 @@ func main() {
 			"CONTENT_TOPIC":  *contentTopic,
 			"METADATA_TOPIC": *metadataTopic,
 			"GROUP_ID":       *consumerGroupID,
-			"KAFKA_ADDRS":    *consumerAddrs,
+			"KAFKA_ADDRESS":  *consumerAddress,
+			"LAG_TOLERANCE":  *consumerLagTolerance,
 			"E2E_TEST_IDS":   *e2eTestUUIDs,
 		}).Infof("[Startup] notifications-push is starting ")
 
@@ -172,16 +180,7 @@ func main() {
 			kafkaTopics = append(kafkaTopics, *metadataTopic)
 		}
 
-		kafkaConsumer, err := createSupervisedConsumer(log,
-			*consumerAddrs,
-			*consumerGroupID,
-			kafkaTopics,
-			serviceName,
-		)
-
-		if err != nil {
-			log.WithError(err).Fatal("could not start kafka consumer")
-		}
+		kafkaConsumer := createConsumer(log, *consumerAddress, *consumerGroupID, kafkaTopics, *consumerLagTolerance)
 
 		httpClient := &http.Client{
 			Transport: &http.Transport{

app_integration_test.go

@@ -1,21 +1,20 @@
+//go:build integration
 // +build integration
 
 package main
 
 import (
 	"bytes"
 	"context"
-	"io/ioutil"
+	"fmt"
 	"net/http"
 	"net/http/httptest"
 	"regexp"
 	"testing"
 	"time"
 
-	"errors"
-
 	"github.com/Financial-Times/go-logger/v2"
-	"github.com/Financial-Times/kafka-client-go/kafka"
+	"github.com/Financial-Times/kafka-client-go/v3"
 	"github.com/Financial-Times/notifications-push/v5/consumer"
 	"github.com/Financial-Times/notifications-push/v5/dispatch"
 	"github.com/Financial-Times/notifications-push/v5/mocks"
@@ -65,8 +64,6 @@ func TestPushNotifications(t *testing.T) {
 	t.Parallel()
 
 	l := logger.NewUPPLogger("TEST", "info")
-	l.Out = ioutil.Discard
-
 	// handlers vars
 	var (
 		apiGatewayValidateURL = "/api-gateway/validate"
@@ -99,9 +96,6 @@ func TestPushNotifications(t *testing.T) {
 	d, h := startDispatcher(delay, historySize, l)
 	defer d.Stop()
 
-	// consumer
-	msgQueue := createMsgQueue(t, uriWhitelist, typeWhitelist, originWhitelist, resource, "test-api", d, l)
-
 	// server
 	router := mux.NewRouter()
 	server := httptest.NewServer(router)
@@ -145,13 +139,19 @@ func TestPushNotifications(t *testing.T) {
 			invalidContentTypeMsg,
 			annotationMsg,
 		}
+		var buff bytes.Buffer
+		logger := logger.NewUnstructuredLogger()
+		logger.SetOutput(&buff)
+		queue := createMsgQueue(t, uriWhitelist, typeWhitelist, originWhitelist, resource, "test-api", d, logger)
 		for {
 			select {
 			case <-ctx.Done():
 				return
 			case <-time.After(sendDelay):
 				for _, msg := range msgs {
-					if !assert.NoError(t, msgQueue.HandleMessage(msg)) {
+					queue.HandleMessage(msg)
+					buffMessage := buff.String()
+					if !assert.NotContains(t, buffMessage, "error") {
 						return
 					}
 				}
@@ -190,15 +190,15 @@ func testHealthcheckEndpoints(ctx context.Context, t *testing.T, serverURL strin
 				return http.StatusOK, nil
 			},
 			kafkaFunc: func() error {
-				return errors.New("sample error")
+				return fmt.Errorf("error connecting to kafka queue")
 			},
 			expectedStatus: http.StatusServiceUnavailable,
 			expectedBody:   "error connecting to kafka queue",
 		},
 		"gtg endpoint ApiGateway failure": {
 			url: "/__gtg",
 			clientFunc: func(ctx context.Context, url string) (int, error) {
-				return http.StatusServiceUnavailable, errors.New("gateway failed")
+				return http.StatusServiceUnavailable, fmt.Errorf("gateway failed")
 			},
 			kafkaFunc: func() error {
 				return nil

consumer/consume.go

@@ -1,13 +1,13 @@
 package consumer
 
 import (
-	"github.com/Financial-Times/kafka-client-go/kafka"
+	"github.com/Financial-Times/kafka-client-go/v3"
 	"github.com/Financial-Times/notifications-push/v5/dispatch"
 )
 
 // MessageQueueHandler is a generic interface for implementation of components to handle messages form the kafka queue.
 type MessageQueueHandler interface {
-	HandleMessage(queueMsg kafka.FTMessage) error
+	HandleMessage(queueMsg kafka.FTMessage)
 }
 
 type notificationDispatcher interface {
@@ -26,11 +26,12 @@ func NewMessageQueueHandler(contentHandler, metadataHandler MessageQueueHandler)
 	}
 }
 
-func (h *MessageQueueRouter) HandleMessage(queueMsg kafka.FTMessage) error {
+func (h *MessageQueueRouter) HandleMessage(queueMsg kafka.FTMessage) {
 	if h.metadataHandler != nil && isAnnotationMessage(queueMsg.Headers) {
-		return h.metadataHandler.HandleMessage(queueMsg)
+		h.metadataHandler.HandleMessage(queueMsg)
+		return
 	}
-	return h.contentHandler.HandleMessage(queueMsg)
+	h.contentHandler.HandleMessage(queueMsg)
 }
 
 func isAnnotationMessage(msgHeaders map[string]string) bool {

consumer/content.go

@@ -5,7 +5,7 @@ import (
 	"strings"
 
 	"github.com/Financial-Times/go-logger/v2"
-	"github.com/Financial-Times/kafka-client-go/kafka"
+	"github.com/Financial-Times/kafka-client-go/v3"
 )
 
 var exists = struct{}{}
@@ -50,7 +50,7 @@ func NewContentQueueHandler(contentURIWhitelist *regexp.Regexp, contentTypeWhite
 	}
 }
 
-func (qHandler *ContentQueueHandler) HandleMessage(queueMsg kafka.FTMessage) error {
+func (qHandler *ContentQueueHandler) HandleMessage(queueMsg kafka.FTMessage) {
 	msg := NotificationQueueMessage{queueMsg}
 	tid := msg.TransactionID()
 	pubEvent, err := msg.AsContent()
@@ -59,31 +59,31 @@ func (qHandler *ContentQueueHandler) HandleMessage(queueMsg kafka.FTMessage) err
 	monitoringLogger := qHandler.log.WithMonitoringEvent("NotificationsPush", tid, contentType)
 	if err != nil {
 		monitoringLogger.WithField("message_body", msg.Body).WithError(err).Warn("Skipping event.")
-		return err
+		return
 	}
 
 	if msg.HasCarouselTransactionID() {
 		monitoringLogger.WithValidFlag(false).WithField("contentUri", pubEvent.ContentURI).Info("Skipping event: Carousel publish event.")
-		return nil
+		return
 	}
 
 	strippedDirectivesContentType := stripDirectives(contentType)
 	isE2ETest := msg.HasE2ETestTransactionID(qHandler.e2eTestUUIDs)
 	if !isE2ETest {
 		if msg.HasSynthTransactionID() {
 			monitoringLogger.WithValidFlag(false).WithField("contentUri", pubEvent.ContentURI).Info("Skipping event: Synthetic transaction ID.")
-			return nil
+			return
 		}
 
 		if strippedDirectivesContentType == "application/json" || strippedDirectivesContentType == "" {
 			if !pubEvent.Matches(qHandler.contentURIWhitelist) {
 				monitoringLogger.WithValidFlag(false).WithField("contentUri", pubEvent.ContentURI).Info("Skipping event: contentUri is not in the whitelist.")
-				return nil
+				return
 			}
 		} else {
 			if !qHandler.contentTypeWhitelist.Contains(strippedDirectivesContentType) {
 				monitoringLogger.WithValidFlag(false).Info("Skipping event: contentType is not the whitelist.")
-				return nil
+				return
 			}
 		}
 	}
@@ -92,7 +92,7 @@ func (qHandler *ContentQueueHandler) HandleMessage(queueMsg kafka.FTMessage) err
 	notification, err := qHandler.mapper.MapNotification(pubEvent, msg.TransactionID())
 	if err != nil {
 		monitoringLogger.WithError(err).Warn("Skipping event: Cannot build notification for message.")
-		return err
+		return
 	}
 	notification.IsE2ETest = isE2ETest
 
@@ -101,8 +101,6 @@ func (qHandler *ContentQueueHandler) HandleMessage(queueMsg kafka.FTMessage) err
 		WithField("notification_type", notification.Type).
 		Info("Valid notification received")
 	qHandler.dispatcher.Send(notification)
-
-	return nil
 }
 
 func stripDirectives(contentType string) string {