Change kafka consumer groups per deployment

Fix healthcheck return values
Financial-Times · Apr 19, 2022 · 79979c0 · 79979c0
1 parent 0a5ee38
commit 79979c0
Show file tree

Hide file tree

Showing 6 changed files with 9 additions and 8 deletions.
diff --git a/helm/notifications-push/app-configs/list-notifications-push_eks_delivery.yaml b/helm/notifications-push/app-configs/list-notifications-push_eks_delivery.yaml
@@ -4,6 +4,7 @@ service:
   name: list-notifications-push
 env:
   TOPIC: "PublicationNotificationEvents"
+  GROUP_ID: "list-notifications-push"
   CONSUMER_BACKOFF: "2"
   NOTIFICATIONS_RESOURCE: "lists"
   CONTENT_TYPE_WHITELIST: "application/vnd.ft-upp-list+json"

diff --git a/helm/notifications-push/app-configs/notifications-push_eks_delivery.yaml b/helm/notifications-push/app-configs/notifications-push_eks_delivery.yaml
@@ -5,6 +5,7 @@ service:
 env:
   TOPIC: "PublicationNotificationEvents"
   METADATA_TOPIC: "PostConceptAnnotations"
+  GROUP_ID: "notifications-push"
   CONSUMER_BACKOFF: "2"
   NOTIFICATIONS_RESOURCE: "content"
   CONTENT_URI_WHITELIST: "^http://(methode|wordpress-article|content|upp)(-collection|-content-placeholder|-notifications-creator)?(-mapper|-unfolder|)(-pr|-iw)?(-uk-.*)?\\.svc\\.ft\\.com(:\\d{2,5})?/(content|complementarycontent)/[\\w-]+.*$"

diff --git a/helm/notifications-push/app-configs/page-notifications-push_eks_delivery.yaml b/helm/notifications-push/app-configs/page-notifications-push_eks_delivery.yaml
@@ -4,6 +4,7 @@ service:
   name: page-notifications-push
 env:
   TOPIC: "PublicationNotificationEvents"
+  GROUP_ID: "page-notifications-push"
   CONSUMER_BACKOFF: "2"
   NOTIFICATIONS_RESOURCE: "pages"
   CONTENT_TYPE_WHITELIST: "application/vnd.ft-upp-page+json"

diff --git a/helm/notifications-push/templates/deployment.yaml b/helm/notifications-push/templates/deployment.yaml
@@ -48,10 +48,7 @@ spec:
               name: global-config
               key: api.host.with.protocol
         - name: GROUP_ID
-          # set this as the pod name so it's unique per instance
-          valueFrom:
-            fieldRef:
-              fieldPath: metadata.name
+          value: {{ .Values.env.GROUP_ID }}
         - name: TOPIC
           value: {{ .Values.env.TOPIC }}
         - name: METADATA_TOPIC

diff --git a/push_service.go b/push_service.go
@@ -25,7 +25,7 @@ type notificationSystem interface {
 func startService(srv *http.Server, n notificationSystem, consumer *kafka.Consumer, msgHandler queueConsumer.MessageQueueHandler, log *logger.UPPLogger) func(time.Duration) {
 	go n.Start()
 
-	consumer.Start(msgHandler.HandleMessage)
+	go consumer.Start(msgHandler.HandleMessage)
 
 	go func() {
 		err := srv.ListenAndServe()

diff --git a/resources/healthchecks.go b/resources/healthchecks.go
@@ -3,6 +3,7 @@ package resources
 import (
 	"context"
 	"errors"
+	"fmt"
 	"net/http"
 	"time"
 
@@ -94,15 +95,15 @@ func (h *HealthCheck) checkKafkaConsumerReachable() (string, error) {
 	if err == nil {
 		return "Connectivity to kafka is OK.", nil
 	}
-	return "Error connecting to kafka", err
+	return "", err
 }
 
 func (h *HealthCheck) checkKafkaConsumerLag() (string, error) {
 	err := h.consumer.MonitorCheck()
 	if err == nil {
 		return "Kafka consumer is not lagging", nil
 	}
-	return "Kafka consumer is lagging behind", err
+	return "", err
 }
 
 // checks if apiGateway service is available
@@ -131,5 +132,5 @@ func (h *HealthCheck) checkAPIGatewayService() (string, error) {
 		return "ApiGateway service is working", nil
 	}
 
-	return "", errors.New("unable to verify ApiGateway service is working")
+	return "", fmt.Errorf("unable to verify ApiGateway service is working")
 }