Add Kafka to processor's integration tests

Author: Daniel Jimenez

cmd/vulnerability-db-consumer/main.go

@@ -132,7 +132,7 @@ func buildKafkaNotifier(conf *config, logger *log.Logger) (*notify.KafkaNotifier
 	if err != nil {
 		return nil, err
 	}
-	return notify.NewKafkaNotifier(&kafkaCli, logger), nil
+	return notify.NewKafkaNotifier(kafkaCli, logger), nil
 }
 
 func setupLogger(cfg config) *log.Logger {

pkg/asyncapi/kafka/client.go

@@ -32,7 +32,7 @@ type Client struct {
 // NewClient creates a new Kafka client connected to the a broker using the
 // given credentials and setting the mapping between all the entities and their
 // corresponding topics.
-func NewClient(user string, password string, broker string, topic string) (Client, error) {
+func NewClient(user string, password string, broker string, topic string) (*Client, error) {
 	config := kafka.ConfigMap{
 		"bootstrap.servers": broker,
 	}
@@ -44,9 +44,9 @@ func NewClient(user string, password string, broker string, topic string) (Clien
 	}
 	p, err := kafka.NewProducer(&config)
 	if err != nil {
-		return Client{}, err
+		return &Client{}, err
 	}
-	return Client{p, topic}, nil
+	return &Client{p, topic}, nil
 }
 
 // Push sends the payload of an entity, with the specified id, to corresponding

test/docker-compose.yml

@@ -3,6 +3,7 @@
 version: '2'
 
 services:
+    # PostgreSQL
     vulndb_test:
         container_name: vulndb_test
         image: postgres:11-alpine
@@ -17,3 +18,22 @@ services:
             - POSTGRES_USER=vulndb_test
             - POSTGRES_PASSWORD=vulndb_test
             - POSTGRES_DB=vulndb_test
+    # Kafka
+    zookeeper:
+        image: confluentinc/cp-zookeeper:7.2.1
+        environment:
+            - ZOOKEEPER_CLIENT_PORT=2181
+            - ZOOKEEPER_TICK_TIME=2000
+    kafka:
+        image: confluentinc/cp-kafka:7.2.1
+        depends_on:
+        - zookeeper
+        ports:
+        - 29092:29092
+        environment:
+            - KAFKA_BROKER_ID=1
+            - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
+            - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:29092
+            - KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
+            - KAFKA_INTER_BROKER_LISTENER_NAME=PLAINTEXT
+            - KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1