Skip to content

Commit 523ff27

Browse files
sakulalijmsnll
authored andcommitted
[receiver/kafkametrics] Using unique container networks and container names and attempt to fix flaky tests (open-telemetry#28903)
**Description:** <Describe what has changed.> Using unique container networks and container names and attempt to fix flaky tests **Link to tracking Issue:** open-telemetry#26293 **Testing:** **Preparation:** DIR = receiver/kafkametricsreceiver CMD = go test -v -count=1 -race -timeout 360s -parallel 4 -tags=integration,"" -run=Integration ./... **Tests:** 1. If we manually modify the code(as shown below) and use invalid kafka broker, such as `localhost:invalid-port`, the same error as shown in the issue may occur. ``` // receiver/kafkametricsreceiver/integration_test.go scraperinttest.WithCustomConfig( func(t *testing.T, cfg component.Config, ci *scraperinttest.ContainerInfo) { rCfg := cfg.(*Config) rCfg.CollectionInterval = 5 * time.Second rCfg.Brokers = []string{"localhost:invalid-port"} rCfg.Scrapers = []string{"brokers", "consumers", "topics"} }), ``` 2. If we execute the test commands **sequentially** , it seems that the execution results are all correct. ``` # all result are correct for i in {1..100}; do echo "Run $i"; ./${CMD} ; done ``` 3. If we execute the commands in **parallel** end with **`&`**, sometimes the error shown in the issue may occur. ``` # sometimes result occur error for i in {1..20}; do echo "Run $i"; ./${CMD} &; done ``` **Inference:** I have found that duplicate container networks and container names can cause container creation to fail or result in successfully created containers with unavailable ports, which may lead to issues similar to the one shown. **Additional information:** Since Kafka's startup relies on ZooKeeper (which waits for the default `zookeeper.connection.timeout.ms=18000`), if Kafka starts first and ZooKeeper fails to start properly after the timeout duration, it will cause the Kafka container to fail to start correctly. I found the issue testcontainers/testcontainers-go#1791 wants to support that. **Documentation:** --------- Signed-off-by: sakulali <sakulali@126.com>
1 parent 9839efa commit 523ff27

File tree

2 files changed

+13
-9
lines changed

2 files changed

+13
-9
lines changed

receiver/kafkametricsreceiver/go.mod

+1-1
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ go 1.20
55
require (
66
github.com/IBM/sarama v1.41.3
77
github.com/google/go-cmp v0.6.0
8+
github.com/google/uuid v1.4.0
89
github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal v0.88.0
910
github.com/open-telemetry/opentelemetry-collector-contrib/internal/kafka v0.88.0
1011
github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest v0.88.0
@@ -44,7 +45,6 @@ require (
4445
github.com/gogo/protobuf v1.3.2 // indirect
4546
github.com/golang/protobuf v1.5.3 // indirect
4647
github.com/golang/snappy v0.0.4 // indirect
47-
github.com/google/uuid v1.4.0 // indirect
4848
github.com/hashicorp/errwrap v1.1.0 // indirect
4949
github.com/hashicorp/go-multierror v1.1.1 // indirect
5050
github.com/hashicorp/go-uuid v1.0.3 // indirect

receiver/kafkametricsreceiver/integration_test.go

+12-8
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@ import (
1111
"testing"
1212
"time"
1313

14+
"github.com/google/uuid"
1415
"github.com/testcontainers/testcontainers-go"
1516
"github.com/testcontainers/testcontainers-go/wait"
1617
"go.opentelemetry.io/collector/component"
@@ -20,13 +21,16 @@ import (
2021
)
2122

2223
const (
23-
networkName = "kafka-network"
2424
kafkaPort = "9092"
2525
zookeeperPort = "2181"
26-
zookeeperHost = "zookeeper"
2726
)
2827

2928
func TestIntegration(t *testing.T) {
29+
uid := fmt.Sprintf("-%s", uuid.NewString())
30+
networkName := "kafka-network" + uid
31+
zkContainerName := "zookeeper" + uid
32+
kafkaContainerName := "kafka" + uid
33+
3034
scraperinttest.NewIntegrationTest(
3135
NewFactory(),
3236
scraperinttest.WithNetworkRequest(
@@ -37,23 +41,23 @@ func TestIntegration(t *testing.T) {
3741
),
3842
scraperinttest.WithContainerRequest(
3943
testcontainers.ContainerRequest{
40-
Name: "zookeeper",
44+
Name: zkContainerName,
4145
Image: "ubuntu/zookeeper:3.1-22.04_beta",
4246
Networks: []string{networkName},
43-
Hostname: zookeeperHost,
47+
Hostname: zkContainerName,
4448
ExposedPorts: []string{zookeeperPort},
4549
WaitingFor: wait.ForAll(
4650
wait.ForListeningPort(zookeeperPort).WithStartupTimeout(2 * time.Minute),
4751
),
4852
}),
4953
scraperinttest.WithContainerRequest(
5054
testcontainers.ContainerRequest{
51-
Name: "kafka",
55+
Name: kafkaContainerName,
5256
Image: "ubuntu/kafka:3.1-22.04_beta",
5357
Networks: []string{networkName},
5458
ExposedPorts: []string{kafkaPort},
5559
Env: map[string]string{
56-
"ZOOKEEPER_HOST": zookeeperHost,
60+
"ZOOKEEPER_HOST": zkContainerName,
5761
"ZOOKEEPER_PORT": zookeeperPort,
5862
},
5963
WaitingFor: wait.ForAll(
@@ -65,8 +69,8 @@ func TestIntegration(t *testing.T) {
6569
rCfg := cfg.(*Config)
6670
rCfg.CollectionInterval = 5 * time.Second
6771
rCfg.Brokers = []string{fmt.Sprintf("%s:%s",
68-
ci.HostForNamedContainer(t, "kafka"),
69-
ci.MappedPortForNamedContainer(t, "kafka", kafkaPort))}
72+
ci.HostForNamedContainer(t, kafkaContainerName),
73+
ci.MappedPortForNamedContainer(t, kafkaContainerName, kafkaPort))}
7074
rCfg.Scrapers = []string{"brokers", "consumers", "topics"}
7175
}),
7276
// scraperinttest.WriteExpected(), // TODO remove

0 commit comments

Comments
 (0)