Fix test

Financial-Times · Oct 26, 2021 · b1d57b6 · b1d57b6
1 parent 1935fbd
commit b1d57b6
Show file tree

Hide file tree

Showing 3 changed files with 5 additions and 7 deletions.
diff --git a/.circleci/config.yml b/.circleci/config.yml
@@ -17,7 +17,7 @@ jobs:
         environment:
           KAFKA_BROKER_ID: 1
           KAFKA_ZOOKEEPER_CONNECT: localhost:2181
-          KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:29092
+          KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost: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

diff --git a/consumer_test.go b/consumer_test.go
@@ -43,7 +43,8 @@ func TestConsumer_InvalidConnection(t *testing.T) {
 			Topics:                  []string{testTopic},
 			Options:                 nil,
 		},
-		logger: log,
+		consumerGroupLock: &sync.RWMutex{},
+		logger:            log,
 	}
 
 	assert.Error(t, consumer.ConnectivityCheck())
@@ -64,13 +65,13 @@ func TestKafkaConsumer_StartListening(t *testing.T) {
 
 	consumer := NewConsumer(config, log, time.Second)
 
-	consumer.StartListening(func(msg FTMessage) {})
+	go consumer.StartListening(func(msg FTMessage) {})
 
 	time.Sleep(3 * time.Second)
 
 	assert.NoError(t, consumer.ConnectivityCheck())
 
-	time.Sleep(1 * time.Second)
+	time.Sleep(3 * time.Second)
 
 	assert.NoError(t, consumer.Close())
 }

diff --git a/producer_test.go b/producer_test.go
@@ -42,9 +42,6 @@ func TestProducer_SendMessage(t *testing.T) {
 	}
 	assert.NoError(t, producer.SendMessage(msg))
 
-	msg = NewFTMessage(nil, `{"foo":"bar"}`)
-	assert.NoError(t, producer.SendMessage(msg))
-
 	assert.NoError(t, producer.Close())
 }