ONSdigital · lindenmckenzie · Jul 27, 2023 · Jun 20, 2023 · Jun 22, 2023 · Jun 22, 2023
diff --git a/Dockerfile.local b/Dockerfile.local
@@ -1,4 +1,4 @@
-FROM golang:1.18-stretch AS base
+FROM golang:1.20.5-bullseye as build
 
 ENV GOCACHE=/go/.go/cache GOPATH=/go/.go/path TZ=Europe/London
 

diff --git a/README.md b/README.md
diff --git a/config/config.go b/config/config.go
@@ -18,36 +18,38 @@ type MongoConfig struct {
 
 // Configuration structure which hold information for configuring the import API
 type Configuration struct {
-	BindAddr                   string        `envconfig:"BIND_ADDR"`
-	KafkaAddr                  []string      `envconfig:"KAFKA_ADDR"                       json:"-"`
-	KafkaSecProtocol           string        `envconfig:"KAFKA_SEC_PROTO"`
-	KafkaSecCACerts            string        `envconfig:"KAFKA_SEC_CA_CERTS"`
-	KafkaSecClientCert         string        `envconfig:"KAFKA_SEC_CLIENT_CERT"`
-	KafkaSecClientKey          string        `envconfig:"KAFKA_SEC_CLIENT_KEY"             json:"-"`
-	KafkaSecSkipVerify         bool          `envconfig:"KAFKA_SEC_SKIP_VERIFY"`
-	GenerateDownloadsTopic     string        `envconfig:"GENERATE_DOWNLOADS_TOPIC"`
-	CantabularExportStartTopic string        `envconfig:"CANTABULAR_EXPORT_START"`
-	CodeListAPIURL             string        `envconfig:"CODE_LIST_API_URL"`
-	DatasetAPIURL              string        `envconfig:"DATASET_API_URL"`
-	WebsiteURL                 string        `envconfig:"WEBSITE_URL"`
-	ZebedeeURL                 string        `envconfig:"ZEBEDEE_URL"`
-	DownloadServiceSecretKey   string        `envconfig:"DOWNLOAD_SERVICE_SECRET_KEY"      json:"-"`
-	ServiceAuthToken           string        `envconfig:"SERVICE_AUTH_TOKEN"               json:"-"`
-	GracefulShutdownTimeout    time.Duration `envconfig:"GRACEFUL_SHUTDOWN_TIMEOUT"`
-	HealthCheckInterval        time.Duration `envconfig:"HEALTHCHECK_INTERVAL"`
-	HealthCheckCriticalTimeout time.Duration `envconfig:"HEALTHCHECK_CRITICAL_TIMEOUT"`
-	EnablePrivateEndpoints     bool          `envconfig:"ENABLE_PRIVATE_ENDPOINTS"`
-	EnableDetachDataset        bool          `envconfig:"ENABLE_DETACH_DATASET"`
-	EnablePermissionsAuth      bool          `envconfig:"ENABLE_PERMISSIONS_AUTH"`
-	EnableObservationEndpoint  bool          `envconfig:"ENABLE_OBSERVATION_ENDPOINT"`
-	DisableGraphDBDependency   bool          `envconfig:"DISABLE_GRAPH_DB_DEPENDENCY"`
-	KafkaVersion               string        `envconfig:"KAFKA_VERSION"`
-	DefaultMaxLimit            int           `envconfig:"DEFAULT_MAXIMUM_LIMIT"`
-	DefaultLimit               int           `envconfig:"DEFAULT_LIMIT"`
-	DefaultOffset              int           `envconfig:"DEFAULT_OFFSET"`
-	MaxRequestOptions          int           `envconfig:"MAX_REQUEST_OPTIONS"`
-	EncryptionDisabled         bool          `envconfig:"ENCRYPTION_DISABLED"`
-	ComponentTestUseLogFile    bool          `envconfig:"COMPONENT_TEST_USE_LOG_FILE"`
+	BindAddr                       string        `envconfig:"BIND_ADDR"`
+	KafkaAddr                      []string      `envconfig:"KAFKA_ADDR"                       json:"-"`
+	KafkaConsumerMinBrokersHealthy int           `envconfig:"KAFKA_CONSUMER_MIN_BROKERS_HEALTHY"`
+	KafkaProducerMinBrokersHealthy int           `envconfig:"KAFKA_PRODUCER_MIN_BROKERS_HEALTHY"`
+	KafkaSecProtocol               string        `envconfig:"KAFKA_SEC_PROTO"`
+	KafkaSecCACerts                string        `envconfig:"KAFKA_SEC_CA_CERTS"`
+	KafkaSecClientCert             string        `envconfig:"KAFKA_SEC_CLIENT_CERT"`
+	KafkaSecClientKey              string        `envconfig:"KAFKA_SEC_CLIENT_KEY"             json:"-"`
+	KafkaSecSkipVerify             bool          `envconfig:"KAFKA_SEC_SKIP_VERIFY"`
+	GenerateDownloadsTopic         string        `envconfig:"GENERATE_DOWNLOADS_TOPIC"`
+	CantabularExportStartTopic     string        `envconfig:"CANTABULAR_EXPORT_START"`
+	CodeListAPIURL                 string        `envconfig:"CODE_LIST_API_URL"`
+	DatasetAPIURL                  string        `envconfig:"DATASET_API_URL"`
+	WebsiteURL                     string        `envconfig:"WEBSITE_URL"`
+	ZebedeeURL                     string        `envconfig:"ZEBEDEE_URL"`
+	DownloadServiceSecretKey       string        `envconfig:"DOWNLOAD_SERVICE_SECRET_KEY"      json:"-"`
+	ServiceAuthToken               string        `envconfig:"SERVICE_AUTH_TOKEN"               json:"-"`
+	GracefulShutdownTimeout        time.Duration `envconfig:"GRACEFUL_SHUTDOWN_TIMEOUT"`
+	HealthCheckInterval            time.Duration `envconfig:"HEALTHCHECK_INTERVAL"`
+	HealthCheckCriticalTimeout     time.Duration `envconfig:"HEALTHCHECK_CRITICAL_TIMEOUT"`
+	EnablePrivateEndpoints         bool          `envconfig:"ENABLE_PRIVATE_ENDPOINTS"`
+	EnableDetachDataset            bool          `envconfig:"ENABLE_DETACH_DATASET"`
+	EnablePermissionsAuth          bool          `envconfig:"ENABLE_PERMISSIONS_AUTH"`
+	EnableObservationEndpoint      bool          `envconfig:"ENABLE_OBSERVATION_ENDPOINT"`
+	DisableGraphDBDependency       bool          `envconfig:"DISABLE_GRAPH_DB_DEPENDENCY"`
+	KafkaVersion                   string        `envconfig:"KAFKA_VERSION"`
+	DefaultMaxLimit                int           `envconfig:"DEFAULT_MAXIMUM_LIMIT"`
+	DefaultLimit                   int           `envconfig:"DEFAULT_LIMIT"`
+	DefaultOffset                  int           `envconfig:"DEFAULT_OFFSET"`
+	MaxRequestOptions              int           `envconfig:"MAX_REQUEST_OPTIONS"`
+	EncryptionDisabled             bool          `envconfig:"ENCRYPTION_DISABLED"`
+	ComponentTestUseLogFile        bool          `envconfig:"COMPONENT_TEST_USE_LOG_FILE"`
 	MongoConfig
 }
 
@@ -69,29 +71,31 @@ func Get() (*Configuration, error) {
 	}
 
 	cfg = &Configuration{
-		BindAddr:                   ":22000",
-		KafkaAddr:                  []string{"localhost:9092", "localhost:9093", "localhost:9094"},
-		GenerateDownloadsTopic:     "filter-job-submitted",
-		CantabularExportStartTopic: "cantabular-export-start",
-		CodeListAPIURL:             "http://localhost:22400",
-		DatasetAPIURL:              "http://localhost:22000",
-		WebsiteURL:                 "http://localhost:20000",
-		ZebedeeURL:                 "http://localhost:8082",
-		ServiceAuthToken:           "FD0108EA-825D-411C-9B1D-41EF7727F465",
-		DownloadServiceSecretKey:   "QB0108EZ-825D-412C-9B1D-41EF7747F462",
-		GracefulShutdownTimeout:    5 * time.Second,
-		HealthCheckInterval:        30 * time.Second,
-		HealthCheckCriticalTimeout: 90 * time.Second,
-		EnablePrivateEndpoints:     false,
-		EnableDetachDataset:        false,
-		EnablePermissionsAuth:      false,
-		EnableObservationEndpoint:  true,
-		DisableGraphDBDependency:   false,
-		KafkaVersion:               "1.0.2",
-		DefaultMaxLimit:            1000,
-		DefaultLimit:               20,
-		DefaultOffset:              0,
-		MaxRequestOptions:          100, // Maximum number of options acceptable in an incoming Patch request. Compromise between one option per call (inefficient) and an order of 100k options per call, for census data (memory and computationally expensive)
+		BindAddr:                       ":22000",
+		KafkaAddr:                      []string{"localhost:9092", "localhost:9093", "localhost:9094"},
+		KafkaConsumerMinBrokersHealthy: 1,
+		KafkaProducerMinBrokersHealthy: 2,
+		GenerateDownloadsTopic:         "filter-job-submitted",
+		CantabularExportStartTopic:     "cantabular-export-start",
+		CodeListAPIURL:                 "http://localhost:22400",
+		DatasetAPIURL:                  "http://localhost:22000",
+		WebsiteURL:                     "http://localhost:20000",
+		ZebedeeURL:                     "http://localhost:8082",
+		ServiceAuthToken:               "FD0108EA-825D-411C-9B1D-41EF7727F465",
+		DownloadServiceSecretKey:       "QB0108EZ-825D-412C-9B1D-41EF7747F462",
+		GracefulShutdownTimeout:        5 * time.Second,
+		HealthCheckInterval:            30 * time.Second,
+		HealthCheckCriticalTimeout:     90 * time.Second,
+		EnablePrivateEndpoints:         false,
+		EnableDetachDataset:            false,
+		EnablePermissionsAuth:          false,
+		EnableObservationEndpoint:      true,
+		DisableGraphDBDependency:       false,
+		KafkaVersion:                   "1.0.2",
+		DefaultMaxLimit:                1000,
+		DefaultLimit:                   20,
+		DefaultOffset:                  0,
+		MaxRequestOptions:              100, // Maximum number of options acceptable in an incoming Patch request. Compromise between one option per call (inefficient) and an order of 100k options per call, for census data (memory and computationally expensive)
 		MongoConfig: MongoConfig{
 			MongoDriverConfig: mongodriver.MongoDriverConfig{
 				ClusterEndpoint:               "localhost:27017",

diff --git a/config/config_test.go b/config/config_test.go
@@ -22,6 +22,8 @@ func TestSpec(t *testing.T) {
 			Convey("The values should be set to the expected defaults", func() {
 				So(cfg.BindAddr, ShouldEqual, ":22000")
 				So(cfg.KafkaAddr, ShouldResemble, []string{"localhost:9092", "localhost:9093", "localhost:9094"})
+				So(cfg.KafkaConsumerMinBrokersHealthy, ShouldEqual, 1)
+				So(cfg.KafkaProducerMinBrokersHealthy, ShouldEqual, 2)
 				So(cfg.KafkaVersion, ShouldEqual, "1.0.2")
 				So(cfg.KafkaSecProtocol, ShouldEqual, "")
 				So(cfg.KafkaSecClientCert, ShouldEqual, "")

diff --git a/features/compose/dp-dataset-api.yml b/features/compose/dp-dataset-api.yml
@@ -23,3 +23,4 @@ services:
       SERVICE_AUTH_TOKEN: "testToken"
       COMPONENT_TEST_USE_LOG_FILE: "${COMPONENT_TEST_USE_LOG_FILE-false}"
       KAFKA_SEC_PROTO: ""
+      KAFKA_PRODUCER_MIN_BROKERS_HEALTHY: 1
diff --git a/features/steps/dataset_component.go b/features/steps/dataset_component.go
@@ -14,8 +14,8 @@ import (
 	"github.com/ONSdigital/dp-dataset-api/store"
 	storeMock "github.com/ONSdigital/dp-dataset-api/store/datastoretest"
 	"github.com/ONSdigital/dp-healthcheck/healthcheck"
-	kafka "github.com/ONSdigital/dp-kafka/v2"
-	"github.com/ONSdigital/dp-kafka/v2/kafkatest"
+	kafka "github.com/ONSdigital/dp-kafka/v3"
+	"github.com/ONSdigital/dp-kafka/v3/kafkatest"
 	mongodriver "github.com/ONSdigital/dp-mongodb/v3/mongodb"
 	"github.com/ONSdigital/log.go/v2/log"
 )
@@ -132,21 +132,35 @@ func funcClose(_ context.Context) error {
 }
 
 func (c *DatasetComponent) setConsumer(topic string) error {
+
+	ctx := context.Background()
+
 	var err error
 	kafkaOffset := kafka.OffsetOldest
 	if c.consumer, err = kafka.NewConsumerGroup(
-		context.Background(),
-		c.Config.KafkaAddr,
-		topic,
-		"test-kafka-group",
-		kafka.CreateConsumerGroupChannels(1),
+		ctx,
 		&kafka.ConsumerGroupConfig{
-			Offset:       &kafkaOffset,
-			KafkaVersion: &c.Config.KafkaVersion,
+			BrokerAddrs:       c.Config.KafkaAddr,
+			Topic:             topic,
+			Offset:            &kafkaOffset,
+			KafkaVersion:      &c.Config.KafkaVersion,
+			GroupName:         "test-kafka-group",
+			MinBrokersHealthy: &c.Config.KafkaConsumerMinBrokersHealthy,
 		},
 	); err != nil {
 		return fmt.Errorf("error creating kafka consumer: %w", err)
 	}
+
+	// start consumer group
+	if err := c.consumer.Start(); err != nil {
+		return fmt.Errorf("error starting kafka consumer: %w", err)
+	}
+
+	c.consumer.LogErrors(ctx)
+
+	c.consumer.StateWait(kafka.Consuming)
+	log.Info(context.Background(), "component-test kafka consumer is in consuming state")
+
 	return nil
 }
 
@@ -166,7 +180,10 @@ func (c *DatasetComponent) DoGetHTTPServer(bindAddr string, router http.Handler)
 
 func (c *DatasetComponent) DoGetKafkaProducer(ctx context.Context, cfg *config.Configuration, topic string) (kafka.IProducer, error) {
 	pConfig := &kafka.ProducerConfig{
-		KafkaVersion: &cfg.KafkaVersion,
+		KafkaVersion:      &cfg.KafkaVersion,
+		Topic:             topic,
+		BrokerAddrs:       cfg.KafkaAddr,
+		MinBrokersHealthy: &cfg.KafkaProducerMinBrokersHealthy,
 	}
 
 	if cfg.KafkaSecProtocol == "TLS" {
@@ -178,8 +195,7 @@ func (c *DatasetComponent) DoGetKafkaProducer(ctx context.Context, cfg *config.C
 		)
 	}
 
-	pChannels := kafka.CreateProducerChannels()
-	return kafka.NewProducer(ctx, cfg.KafkaAddr, topic, pChannels, pConfig)
+	return kafka.NewProducer(ctx, pConfig)
 }
 
 func (c *DatasetComponent) DoGetMockedKafkaProducerOk(_ context.Context, _ *config.Configuration, _ string) (kafka.IProducer, error) {
@@ -188,6 +204,9 @@ func (c *DatasetComponent) DoGetMockedKafkaProducerOk(_ context.Context, _ *conf
 			return &kafka.ProducerChannels{}
 		},
 		CloseFunc: funcClose,
+		LogErrorsFunc: func(ctx context.Context) {
+			// Do nothing
+		},
 	}, nil
 }
 

diff --git a/features/steps/steps.go b/features/steps/steps.go
@@ -131,8 +131,12 @@ func (c *DatasetComponent) theVersionInTheDatabaseForIdShouldBe(id string, body
 }
 
 func (c *DatasetComponent) iHaveARealKafkaContainerWithTopic(topic string) error {
-	c.setConsumer(topic)
+	err := c.setConsumer(topic)
+	if err != nil {
+		return fmt.Errorf("failed to create kafka consumer: %w", err)
+	}
 	c.setInitialiserRealKafka()
+
 	return nil
 }
 

diff --git a/go.mod b/go.mod
@@ -7,31 +7,35 @@ go 1.20
 //     - sonatype-2021-1401 # pkg:golang/github.com/miekg/[email protected]
 replace github.com/spf13/cobra => github.com/spf13/cobra v1.7.0
 
+// to avoid the following vulnerabilities:
+//     - CVE-2023-32731 # pkg:google.golang.org/grpc
+replace google.golang.org/grpc => google.golang.org/grpc v1.55.0
+
 // to avoid 'sonatype-2021-4899' non-CVE Vulnerability
 exclude github.com/gorilla/sessions v1.2.1
 
 require (
-	github.com/ONSdigital/dp-api-clients-go/v2 v2.252.1
+	github.com/ONSdigital/dp-api-clients-go/v2 v2.254.0
 	github.com/ONSdigital/dp-assistdog v0.0.1
 	github.com/ONSdigital/dp-authorisation v0.2.1
 	github.com/ONSdigital/dp-component-test v0.9.2
-	github.com/ONSdigital/dp-graph/v2 v2.16.0
+	github.com/ONSdigital/dp-graph/v2 v2.16.1
 	github.com/ONSdigital/dp-healthcheck v1.6.1
-	github.com/ONSdigital/dp-kafka/v2 v2.8.0
+	github.com/ONSdigital/dp-kafka/v3 v3.10.0
 	github.com/ONSdigital/dp-mongodb/v3 v3.6.0
-	github.com/ONSdigital/dp-net/v2 v2.9.1
+	github.com/ONSdigital/dp-net/v2 v2.10.0
 	github.com/ONSdigital/log.go/v2 v2.4.1
 	github.com/cucumber/godog v0.12.6
-	github.com/google/go-cmp v0.5.8
+	github.com/google/go-cmp v0.5.9
 	github.com/gorilla/mux v1.8.0
 	github.com/jinzhu/copier v0.3.5
 	github.com/justinas/alice v1.2.0
 	github.com/kelseyhightower/envconfig v1.4.0
 	github.com/pkg/errors v0.9.1
 	github.com/satori/go.uuid v1.2.1-0.20181028125025-b2ce2384e17b
 	github.com/smartystreets/goconvey v1.8.0
-	github.com/stretchr/testify v1.8.2
-	go.mongodb.org/mongo-driver v1.11.4
+	github.com/stretchr/testify v1.8.4
+	go.mongodb.org/mongo-driver v1.12.0
 )
 
 require (
@@ -41,8 +45,8 @@ require (
 	github.com/ONSdigital/golang-neo4j-bolt-driver v0.0.0-20210408132126-c2323ff08bf1 // indirect
 	github.com/ONSdigital/graphson v0.3.0 // indirect
 	github.com/ONSdigital/gremgo-neptune v1.0.3 // indirect
-	github.com/Shopify/sarama v1.35.0 // indirect
-	github.com/chromedp/cdproto v0.0.0-20230419194459-b5ff65bc57a3 // indirect
+	github.com/Shopify/sarama v1.38.1 // indirect
+	github.com/chromedp/cdproto v0.0.0-20230620000757-8605e5981815 // indirect
 	github.com/chromedp/chromedp v0.9.1 // indirect
 	github.com/chromedp/sysutil v1.0.0 // indirect
 	github.com/cucumber/gherkin-go/v19 v19.0.3 // indirect
@@ -55,7 +59,7 @@ require (
 	github.com/go-avro/avro v0.0.0-20171219232920-444163702c11 // indirect
 	github.com/gobwas/httphead v0.1.0 // indirect
 	github.com/gobwas/pool v0.2.1 // indirect
-	github.com/gobwas/ws v1.2.0 // indirect
+	github.com/gobwas/ws v1.2.1 // indirect
 	github.com/gofrs/uuid v4.4.0+incompatible // indirect
 	github.com/golang/snappy v0.0.4 // indirect
 	github.com/google/uuid v1.3.0 // indirect
@@ -75,27 +79,28 @@ require (
 	github.com/jcmturner/rpc/v2 v2.0.3 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
 	github.com/jtolds/gls v4.20.0+incompatible // indirect
-	github.com/klauspost/compress v1.16.5 // indirect
+	github.com/klauspost/compress v1.16.6 // indirect
 	github.com/mailru/easyjson v0.7.7 // indirect
 	github.com/mattn/go-colorable v0.1.13 // indirect
-	github.com/mattn/go-isatty v0.0.18 // indirect
+	github.com/mattn/go-isatty v0.0.19 // indirect
 	github.com/maxcnunes/httpfake v1.2.4 // indirect
-	github.com/montanaflynn/stats v0.7.0 // indirect
-	github.com/pierrec/lz4/v4 v4.1.17 // indirect
+	github.com/montanaflynn/stats v0.7.1 // indirect
+	github.com/pierrec/lz4/v4 v4.1.18 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 // indirect
+	github.com/rogpeppe/go-internal v1.10.0 // indirect
 	github.com/smartystreets/assertions v1.13.1 // indirect
 	github.com/spf13/afero v1.9.5 // indirect
 	github.com/spf13/pflag v1.0.5 // indirect
-	github.com/square/mongo-lock v0.0.0-20220601164918-701ecf357cd7 // indirect
+	github.com/square/mongo-lock v0.0.0-20230501160337-eb07183bebe1 // indirect
 	github.com/xdg-go/pbkdf2 v1.0.0 // indirect
 	github.com/xdg-go/scram v1.1.2 // indirect
 	github.com/xdg-go/stringprep v1.0.4 // indirect
 	github.com/youmark/pkcs8 v0.0.0-20201027041543-1326539a0a0a // indirect
-	golang.org/x/crypto v0.8.0 // indirect
-	golang.org/x/net v0.9.0 // indirect
-	golang.org/x/sync v0.1.0 // indirect
-	golang.org/x/sys v0.7.0 // indirect
-	golang.org/x/text v0.9.0 // indirect
+	golang.org/x/crypto v0.10.0 // indirect
+	golang.org/x/net v0.11.0 // indirect
+	golang.org/x/sync v0.3.0 // indirect
+	golang.org/x/sys v0.9.0 // indirect
+	golang.org/x/text v0.10.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )