From 2519d4b51441e3cb7d98977e22b47f153a4adf59 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Manuel=20de=20la=20Pe=C3=B1a?= Date: Tue, 12 Sep 2023 03:31:55 +0200 Subject: [PATCH 01/16] chore: scaffolding for kafka module --- .github/dependabot.yml | 7 + .github/workflows/ci.yml | 2 +- .vscode/.testcontainers-go.code-workspace | 4 + docs/modules/kafka.md | 65 +++++++++ mkdocs.yml | 1 + modules/kafka/Makefile | 5 + modules/kafka/examples_test.go | 36 +++++ modules/kafka/go.mod | 42 ++++++ modules/kafka/go.sum | 160 ++++++++++++++++++++++ modules/kafka/kafka.go | 35 +++++ modules/kafka/kafka_test.go | 24 ++++ 11 files changed, 380 insertions(+), 1 deletion(-) create mode 100644 docs/modules/kafka.md create mode 100644 modules/kafka/Makefile create mode 100644 modules/kafka/examples_test.go create mode 100644 modules/kafka/go.mod create mode 100644 modules/kafka/go.sum create mode 100644 modules/kafka/kafka.go create mode 100644 modules/kafka/kafka_test.go diff --git a/.github/dependabot.yml b/.github/dependabot.yml index ca6786403f..1d45a5c97a 100644 --- a/.github/dependabot.yml +++ b/.github/dependabot.yml @@ -135,6 +135,13 @@ updates: day: sunday open-pull-requests-limit: 3 rebase-strategy: disabled + - package-ecosystem: gomod + directory: /modules/kafka + schedule: + interval: monthly + day: sunday + open-pull-requests-limit: 3 + rebase-strategy: disabled - package-ecosystem: gomod directory: /modules/localstack schedule: diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d9fe9e1208..461049d6a9 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -110,7 +110,7 @@ jobs: matrix: go-version: [1.20.x, 1.x] platform: [ubuntu-latest, macos-latest] - module: [artemis, clickhouse, compose, couchbase, elasticsearch, k3s, localstack, mariadb, mongodb, mysql, nats, neo4j, postgres, pulsar, redis, redpanda, vault] + module: [artemis, clickhouse, compose, couchbase, elasticsearch, k3s, kafka, localstack, mariadb, mongodb, mysql, nats, neo4j, postgres, pulsar, redis, redpanda, vault] uses: ./.github/workflows/ci-test-go.yml secrets: inherit with: diff --git a/.vscode/.testcontainers-go.code-workspace b/.vscode/.testcontainers-go.code-workspace index 5765346528..4b524021bf 100644 --- a/.vscode/.testcontainers-go.code-workspace +++ b/.vscode/.testcontainers-go.code-workspace @@ -65,6 +65,10 @@ "name": "module / k3s", "path": "../modules/k3s" }, + { + "name": "module / kafka", + "path": "../modules/kafka" + }, { "name": "module / localstack", "path": "../modules/localstack" diff --git a/docs/modules/kafka.md b/docs/modules/kafka.md new file mode 100644 index 0000000000..c5085163fe --- /dev/null +++ b/docs/modules/kafka.md @@ -0,0 +1,65 @@ +# Kafka + +Not available until the next release of testcontainers-go :material-tag: main + +## Introduction + +The Testcontainers module for Kafka. + +## Adding this module to your project dependencies + +Please run the following command to add the Kafka module to your Go dependencies: + +``` +go get github.com/testcontainers/testcontainers-go/modules/kafka +``` + +## Usage example + + +[Creating a Kafka container](../../modules/kafka/examples_test.go) inside_block:runKafkaContainer + + +## Module reference + +The Kafka module exposes one entrypoint function to create the Kafka container, and this function receives two parameters: + +```golang +func RunContainer(ctx context.Context, opts ...testcontainers.ContainerCustomizer) (*KafkaContainer, error) +``` + +- `context.Context`, the Go context. +- `testcontainers.ContainerCustomizer`, a variadic argument for passing options. + +### Container Options + +When starting the Kafka container, you can pass options in a variadic way to configure it. + +#### Image + +If you need to set a different Kafka Docker image, you can use `testcontainers.WithImage` with a valid Docker image +for Kafka. E.g. `testcontainers.WithImage("confluentinc/cp-kafka:7.3.3")`. + +#### Wait Strategies + +If you need to set a different wait strategy for Kafka, you can use `testcontainers.WithWaitStrategy` with a valid wait strategy +for Kafka. + +!!!info + The default deadline for the wait strategy is 60 seconds. + +At the same time, it's possible to set a wait strategy and a custom deadline with `testcontainers.WithWaitStrategyAndDeadline`. + +#### Docker type modifiers + +If you need an advanced configuration for Kafka, you can leverage the following Docker type modifiers: + +- `testcontainers.WithConfigModifier` +- `testcontainers.WithHostConfigModifier` +- `testcontainers.WithEndpointSettingsModifier` + +Please read the [Create containers: Advanced Settings](../features/creating_container.md#advanced-settings) documentation for more information. + +### Container Methods + +The Kafka container exposes the following methods: diff --git a/mkdocs.yml b/mkdocs.yml index d0067b6dd9..0e2c62fee1 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -67,6 +67,7 @@ nav: - modules/couchbase.md - modules/elasticsearch.md - modules/k3s.md + - modules/kafka.md - modules/localstack.md - modules/mariadb.md - modules/mongodb.md diff --git a/modules/kafka/Makefile b/modules/kafka/Makefile new file mode 100644 index 0000000000..71ef5f4574 --- /dev/null +++ b/modules/kafka/Makefile @@ -0,0 +1,5 @@ +include ../../commons-test.mk + +.PHONY: test +test: + $(MAKE) test-kafka diff --git a/modules/kafka/examples_test.go b/modules/kafka/examples_test.go new file mode 100644 index 0000000000..91784f610d --- /dev/null +++ b/modules/kafka/examples_test.go @@ -0,0 +1,36 @@ +package kafka_test + +import ( + "context" + "fmt" + + "github.com/testcontainers/testcontainers-go/modules/kafka" +) + +func ExampleRunContainer() { + // runKafkaContainer { + ctx := context.Background() + + kafkaContainer, err := kafka.RunContainer(ctx) + if err != nil { + panic(err) + } + + // Clean up the container after + defer func() { + if err := kafkaContainer.Terminate(ctx); err != nil { + panic(err) + } + }() + // } + + state, err := kafkaContainer.State(ctx) + if err != nil { + panic(err) + } + + fmt.Println(state.Running) + + // Output: + // true +} diff --git a/modules/kafka/go.mod b/modules/kafka/go.mod new file mode 100644 index 0000000000..c7b030a25d --- /dev/null +++ b/modules/kafka/go.mod @@ -0,0 +1,42 @@ +module github.com/testcontainers/testcontainers-go/modules/kafka + +go 1.20 + +require github.com/testcontainers/testcontainers-go v0.23.0 + +require ( + dario.cat/mergo v1.0.0 // indirect + github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1 // indirect + github.com/Microsoft/go-winio v0.6.1 // indirect + github.com/cenkalti/backoff/v4 v4.2.1 // indirect + github.com/containerd/containerd v1.7.3 // indirect + github.com/cpuguy83/dockercfg v0.3.1 // indirect + github.com/docker/distribution v2.8.2+incompatible // indirect + github.com/docker/docker v24.0.6+incompatible // indirect + github.com/docker/go-connections v0.4.0 // indirect + github.com/docker/go-units v0.5.0 // indirect + github.com/gogo/protobuf v1.3.2 // indirect + github.com/golang/protobuf v1.5.3 // indirect + github.com/google/uuid v1.3.1 // indirect + github.com/klauspost/compress v1.16.0 // indirect + github.com/magiconair/properties v1.8.7 // indirect + github.com/moby/patternmatcher v0.5.0 // indirect + github.com/moby/sys/sequential v0.5.0 // indirect + github.com/moby/term v0.5.0 // indirect + github.com/morikuni/aec v1.0.0 // indirect + github.com/opencontainers/go-digest v1.0.0 // indirect + github.com/opencontainers/image-spec v1.1.0-rc4 // indirect + github.com/opencontainers/runc v1.1.5 // indirect + github.com/pkg/errors v0.9.1 // indirect + github.com/sirupsen/logrus v1.9.0 // indirect + golang.org/x/exp v0.0.0-20230510235704-dd950f8aeaea // indirect + golang.org/x/mod v0.9.0 // indirect + golang.org/x/net v0.9.0 // indirect + golang.org/x/sys v0.11.0 // indirect + golang.org/x/tools v0.7.0 // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20230525234030-28d5490b6b19 // indirect + google.golang.org/grpc v1.57.0 // indirect + google.golang.org/protobuf v1.30.0 // indirect +) + +replace github.com/testcontainers/testcontainers-go => ../.. diff --git a/modules/kafka/go.sum b/modules/kafka/go.sum new file mode 100644 index 0000000000..8b6d587397 --- /dev/null +++ b/modules/kafka/go.sum @@ -0,0 +1,160 @@ +dario.cat/mergo v1.0.0 h1:AGCNq9Evsj31mOgNPcLyXc+4PNABt905YmuqPYYpBWk= +dario.cat/mergo v1.0.0/go.mod h1:uNxQE+84aUszobStD9th8a29P2fMDhsBdgRYvZOxGmk= +github.com/AdaLogics/go-fuzz-headers v0.0.0-20230106234847-43070de90fa1 h1:EKPd1INOIyr5hWOWhvpmQpY6tKjeG0hT1s3AMC/9fic= +github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1 h1:UQHMgLO+TxOElx5B5HZ4hJQsoJ/PvUvKRhJHDQXO8P8= +github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1/go.mod h1:xomTg63KZ2rFqZQzSB4Vz2SUXa1BpHTVz9L5PTmPC4E= +github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU= +github.com/Microsoft/go-winio v0.6.1 h1:9/kr64B9VUZrLm5YYwbGtUJnMgqWVOdUAXu6Migciow= +github.com/Microsoft/go-winio v0.6.1/go.mod h1:LRdKpFKfdobln8UmuiYcKPot9D2v6svN5+sAH+4kjUM= +github.com/Microsoft/hcsshim v0.10.0-rc.8 h1:YSZVvlIIDD1UxQpJp0h+dnpLUw+TrY0cx8obKsp3bek= +github.com/cenkalti/backoff/v4 v4.2.1 h1:y4OZtCnogmCPw98Zjyt5a6+QwPLGkiQsYW5oUqylYbM= +github.com/cenkalti/backoff/v4 v4.2.1/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE= +github.com/checkpoint-restore/go-criu/v5 v5.3.0/go.mod h1:E/eQpaFtUKGOOSEBZgmKAcn+zUUwWxqcaKZlF54wK8E= +github.com/cilium/ebpf v0.7.0/go.mod h1:/oI2+1shJiTGAMgl6/RgJr36Eo1jzrRcAWbcXO2usCA= +github.com/containerd/console v1.0.3/go.mod h1:7LqA/THxQ86k76b8c/EMSiaJ3h1eZkMkXar0TQ1gf3U= +github.com/containerd/containerd v1.7.3 h1:cKwYKkP1eTj54bP3wCdXXBymmKRQMrWjkLSWZZJDa8o= +github.com/containerd/containerd v1.7.3/go.mod h1:32FOM4/O0RkNg7AjQj3hDzN9cUGtu+HMvaKUNiqCZB8= +github.com/coreos/go-systemd/v22 v22.3.2/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc= +github.com/cpuguy83/dockercfg v0.3.1 h1:/FpZ+JaygUR/lZP2NlFI2DVfrOEMAIKP5wWEJdoYe9E= +github.com/cpuguy83/dockercfg v0.3.1/go.mod h1:sugsbF4//dDlL/i+S+rtpIWp+5h0BHJHfjj5/jFyUJc= +github.com/cpuguy83/go-md2man/v2 v2.0.0-20190314233015-f79a8a8ca69d/go.mod h1:maD7wRr/U5Z6m/iR4s+kqSMx2CaBsrgA7czyZG/E6dU= +github.com/creack/pty v1.1.18 h1:n56/Zwd5o6whRC5PMGretI4IdRLlmBXYNjScPaBgsbY= +github.com/cyphar/filepath-securejoin v0.2.3 h1:YX6ebbZCZP7VkM3scTTokDgBL2TY741X51MTk3ycuNI= +github.com/cyphar/filepath-securejoin v0.2.3/go.mod h1:aPGpWjXOXUn2NCNjFvBE6aRxGGx79pTxQpKOJNYHHl4= +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/docker/distribution v2.8.2+incompatible h1:T3de5rq0dB1j30rp0sA2rER+m322EBzniBPB6ZIzuh8= +github.com/docker/distribution v2.8.2+incompatible/go.mod h1:J2gT2udsDAN96Uj4KfcMRqY0/ypR+oyYUYmja8H+y+w= +github.com/docker/docker v24.0.6+incompatible h1:hceabKCtUgDqPu+qm0NgsaXf28Ljf4/pWFL7xjWWDgE= +github.com/docker/docker v24.0.6+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk= +github.com/docker/go-connections v0.4.0 h1:El9xVISelRB7BuFusrZozjnkIM5YnzCViNKohAFqRJQ= +github.com/docker/go-connections v0.4.0/go.mod h1:Gbd7IOopHjR8Iph03tsViu4nIes5XhDvyHbTtUxmeec= +github.com/docker/go-units v0.4.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk= +github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4= +github.com/docker/go-units v0.5.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk= +github.com/frankban/quicktest v1.11.3/go.mod h1:wRf/ReqHper53s+kmmSZizM8NamnL3IM0I9ntUbOk+k= +github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA= +github.com/godbus/dbus/v5 v5.0.6/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA= +github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q= +github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q= +github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk= +github.com/golang/protobuf v1.5.3 h1:KhyjKVUg7Usr/dYsdSqoFveMYd5ko72D+zANwlG1mmg= +github.com/golang/protobuf v1.5.3/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY= +github.com/google/go-cmp v0.5.4/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= +github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= +github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38= +github.com/google/uuid v1.3.1 h1:KjJaJ9iWZ3jOFZIf1Lqf4laDRCasjl0BCmnEGxkdLb4= +github.com/google/uuid v1.3.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8= +github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck= +github.com/klauspost/compress v1.16.0 h1:iULayQNOReoYUe+1qtKOqw9CwJv3aNQu8ivo7lw1HU4= +github.com/klauspost/compress v1.16.0/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE= +github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI= +github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= +github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= +github.com/magiconair/properties v1.8.7 h1:IeQXZAiQcpL9mgcAe1Nu6cX9LLw6ExEHKjN0VQdvPDY= +github.com/magiconair/properties v1.8.7/go.mod h1:Dhd985XPs7jluiymwWYZ0G4Z61jb3vdS329zhj2hYo0= +github.com/moby/patternmatcher v0.5.0 h1:YCZgJOeULcxLw1Q+sVR636pmS7sPEn1Qo2iAN6M7DBo= +github.com/moby/patternmatcher v0.5.0/go.mod h1:hDPoyOpDY7OrrMDLaYoY3hf52gNCR/YOUYxkhApJIxc= +github.com/moby/sys/mountinfo v0.5.0/go.mod h1:3bMD3Rg+zkqx8MRYPi7Pyb0Ie97QEBmdxbhnCLlSvSU= +github.com/moby/sys/sequential v0.5.0 h1:OPvI35Lzn9K04PBbCLW0g4LcFAJgHsvXsRyewg5lXtc= +github.com/moby/sys/sequential v0.5.0/go.mod h1:tH2cOOs5V9MlPiXcQzRC+eEyab644PWKGRYaaV5ZZlo= +github.com/moby/term v0.5.0 h1:xt8Q1nalod/v7BqbG21f8mQPqH+xAaC9C3N3wfWbVP0= +github.com/moby/term v0.5.0/go.mod h1:8FzsFHVUBGZdbDsJw/ot+X+d5HLUbvklYLJ9uGfcI3Y= +github.com/morikuni/aec v1.0.0 h1:nP9CBfwrvYnBRgY6qfDQkygYDmYwOilePFkwzv4dU8A= +github.com/morikuni/aec v1.0.0/go.mod h1:BbKIizmSmc5MMPqRYbxO4ZU0S0+P200+tUnFx7PXmsc= +github.com/mrunalp/fileutils v0.5.0/go.mod h1:M1WthSahJixYnrXQl/DFQuteStB1weuxD2QJNHXfbSQ= +github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U= +github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM= +github.com/opencontainers/image-spec v1.1.0-rc4 h1:oOxKUJWnFC4YGHCCMNql1x4YaDfYBTS5Y4x/Cgeo1E0= +github.com/opencontainers/image-spec v1.1.0-rc4/go.mod h1:X4pATf0uXsnn3g5aiGIsVnJBR4mxhKzfwmvK/B2NTm8= +github.com/opencontainers/runc v1.1.5 h1:L44KXEpKmfWDcS02aeGm8QNTFXTo2D+8MYGDIJ/GDEs= +github.com/opencontainers/runc v1.1.5/go.mod h1:1J5XiS+vdZ3wCyZybsuxXZWGrgSr8fFJHLXuG2PsnNg= +github.com/opencontainers/runtime-spec v1.0.3-0.20210326190908-1c3f411f0417/go.mod h1:jwyrGlmzljRJv/Fgzds9SsS/C5hL+LL3ko9hs6T5lQ0= +github.com/opencontainers/selinux v1.10.0/go.mod h1:2i0OySw99QjzBBQByd1Gr9gSjvuho1lHsJxIJ3gGbJI= +github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= +github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM= +github.com/seccomp/libseccomp-golang v0.9.2-0.20220502022130-f33da4d89646/go.mod h1:JA8cRccbGaA1s33RQf7Y1+q9gHmZX1yB/z9WDN1C6fg= +github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeVRXNmyL/1OwPU0+IJeTBvfc= +github.com/sirupsen/logrus v1.8.1/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0= +github.com/sirupsen/logrus v1.9.0 h1:trlNQbNUG3OdDrDil03MCb1H2o9nJ1x4/5LYw7byDE0= +github.com/sirupsen/logrus v1.9.0/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs= +github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk= +github.com/syndtr/gocapability v0.0.0-20200815063812-42c35b437635/go.mod h1:hkRG7XYTFWNJGYcbNJQlaLq0fg1yr4J4t/NcTQtrfww= +github.com/urfave/cli v1.22.1/go.mod h1:Gos4lmkARVdJ6EkW0WaNv/tZAAMe9V7XWyB60NtXRu0= +github.com/vishvananda/netlink v1.1.0/go.mod h1:cTgwzPIzzgDAYoQrMm0EdrjRUBkTqKYppBueQtXaqoE= +github.com/vishvananda/netns v0.0.0-20191106174202-0a2b9b5464df/go.mod h1:JP3t17pCcGlemwknint6hfoeCVQrEMVwxRLRjXpq+BU= +github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= +github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= +golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= +golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= +golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= +golang.org/x/exp v0.0.0-20230510235704-dd950f8aeaea h1:vLCWI/yYrdEHyN2JzIzPO3aaQJHQdp89IZBA/+azVC4= +golang.org/x/exp v0.0.0-20230510235704-dd950f8aeaea/go.mod h1:V1LtkGg67GoY2N1AnLN78QLrzxkLyJw7RJb1gzOOz9w= +golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= +golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= +golang.org/x/mod v0.9.0 h1:KENHtAZL2y3NLMYZeHY9DW8HW8V+kQyJsY/V9JlKvCs= +golang.org/x/mod v0.9.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs= +golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= +golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU= +golang.org/x/net v0.0.0-20201224014010-6772e930b67b/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg= +golang.org/x/net v0.9.0 h1:aWJ/m6xSmxWBx+V0XRHTlrYrPG56jKsLdTFmsSsCzOM= +golang.org/x/net v0.9.0/go.mod h1:d48xBJpPfHeWQsugry2m+kC02ZBRGRgulfHnEXEuWns= +golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.1.0 h1:wsuoTGHzEhffawBOhz5CYhcrV4IdKZbEyZjBMuTp12o= +golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20190606203320-7fc4e5ec1444/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20191026070338-33540a1f6037/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20191115151921-52ab43148777/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20210124154548-22da62e12c0c/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20210616094352-59db8d763f22/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20210906170528-6f6e22806c34/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20211025201205-69cdffdb9359/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20211116061358-0a5406a5449c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.11.0 h1:eG7RXZHdqOJ1i+0lgLgCpSXAp6M3LYlAo6osgSi0xOM= +golang.org/x/sys v0.11.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= +golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= +golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= +golang.org/x/text v0.9.0 h1:2sjJmO8cDvYveuX97RDLsxlyUxLl+GHoLxBiRdHllBE= +golang.org/x/time v0.0.0-20220210224613-90d013bbcef8 h1:vVKdlvoWBphwdxWKrFZEuM0kGgGLxUOYcY4U/2Vjg44= +golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= +golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= +golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE= +golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= +golang.org/x/tools v0.7.0 h1:W4OVu8VVOaIO0yzWMNdepAulS7YfoS3Zabrm8DOXXU4= +golang.org/x/tools v0.7.0/go.mod h1:4pg6aUX35JBAogB10C9AtvVL+qowtN4pT3CGSQex14s= +golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +google.golang.org/genproto/googleapis/rpc v0.0.0-20230525234030-28d5490b6b19 h1:0nDDozoAU19Qb2HwhXadU8OcsiO/09cnTqhUtq2MEOM= +google.golang.org/genproto/googleapis/rpc v0.0.0-20230525234030-28d5490b6b19/go.mod h1:66JfowdXAEgad5O9NnYcsNPLCPZJD++2L9X0PCMODrA= +google.golang.org/grpc v1.57.0 h1:kfzNeI/klCGD2YPMUlaGNT3pxvYfga7smW3Vth8Zsiw= +google.golang.org/grpc v1.57.0/go.mod h1:Sd+9RMTACXwmub0zcNY2c4arhtrbBYD1AUHI/dt16Mo= +google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw= +google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc= +google.golang.org/protobuf v1.27.1/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc= +google.golang.org/protobuf v1.30.0 h1:kPPoIgf3TsEvrm0PFe15JQ+570QVxYzEvvHqChK+cng= +google.golang.org/protobuf v1.30.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gotest.tools/v3 v3.5.0 h1:Ljk6PdHdOhAb5aDMWXjDLMMhph+BpztA4v1QdqEW2eY= diff --git a/modules/kafka/kafka.go b/modules/kafka/kafka.go new file mode 100644 index 0000000000..9eeaf9c657 --- /dev/null +++ b/modules/kafka/kafka.go @@ -0,0 +1,35 @@ +package kafka + +import ( + "context" + + "github.com/testcontainers/testcontainers-go" +) + +// KafkaContainer represents the Kafka container type used in the module +type KafkaContainer struct { + testcontainers.Container +} + +// RunContainer creates an instance of the Kafka container type +func RunContainer(ctx context.Context, opts ...testcontainers.ContainerCustomizer) (*KafkaContainer, error) { + req := testcontainers.ContainerRequest{ + Image: "confluentinc/cp-kafka:7.3.3", + } + + genericContainerReq := testcontainers.GenericContainerRequest{ + ContainerRequest: req, + Started: true, + } + + for _, opt := range opts { + opt.Customize(&genericContainerReq) + } + + container, err := testcontainers.GenericContainer(ctx, genericContainerReq) + if err != nil { + return nil, err + } + + return &KafkaContainer{Container: container}, nil +} diff --git a/modules/kafka/kafka_test.go b/modules/kafka/kafka_test.go new file mode 100644 index 0000000000..61fbec9783 --- /dev/null +++ b/modules/kafka/kafka_test.go @@ -0,0 +1,24 @@ +package kafka + +import ( + "context" + "testing" +) + +func TestKafka(t *testing.T) { + ctx := context.Background() + + container, err := RunContainer(ctx) + if err != nil { + t.Fatal(err) + } + + // Clean up the container after the test is complete + t.Cleanup(func() { + if err := container.Terminate(ctx); err != nil { + t.Fatalf("failed to terminate container: %s", err) + } + }) + + // perform assertions +} From 9ad1c79e4f40721f67250576342b502fbcc7ab58 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Manuel=20de=20la=20Pe=C3=B1a?= Date: Tue, 12 Sep 2023 04:06:01 +0200 Subject: [PATCH 02/16] chore: add test for reading from a topic --- modules/kafka/examples_test.go | 3 +- modules/kafka/go.mod | 27 +++++++++-- modules/kafka/go.sum | 76 ++++++++++++++++++++++++++++--- modules/kafka/kafka.go | 82 +++++++++++++++++++++++++++++++++- modules/kafka/kafka_test.go | 58 +++++++++++++++++++++++- modules/kafka/testing.go | 55 +++++++++++++++++++++++ 6 files changed, 287 insertions(+), 14 deletions(-) create mode 100644 modules/kafka/testing.go diff --git a/modules/kafka/examples_test.go b/modules/kafka/examples_test.go index 91784f610d..39e47e404c 100644 --- a/modules/kafka/examples_test.go +++ b/modules/kafka/examples_test.go @@ -4,6 +4,7 @@ import ( "context" "fmt" + "github.com/testcontainers/testcontainers-go" "github.com/testcontainers/testcontainers-go/modules/kafka" ) @@ -11,7 +12,7 @@ func ExampleRunContainer() { // runKafkaContainer { ctx := context.Background() - kafkaContainer, err := kafka.RunContainer(ctx) + kafkaContainer, err := kafka.RunContainer(ctx, testcontainers.WithImage("confluentinc/cp-kafka:7.3.3")) if err != nil { panic(err) } diff --git a/modules/kafka/go.mod b/modules/kafka/go.mod index c7b030a25d..f61f85af56 100644 --- a/modules/kafka/go.mod +++ b/modules/kafka/go.mod @@ -2,7 +2,11 @@ module github.com/testcontainers/testcontainers-go/modules/kafka go 1.20 -require github.com/testcontainers/testcontainers-go v0.23.0 +require ( + github.com/IBM/sarama v1.41.1 + github.com/docker/go-connections v0.4.0 + github.com/testcontainers/testcontainers-go v0.23.0 +) require ( dario.cat/mergo v1.0.0 // indirect @@ -11,14 +15,26 @@ require ( github.com/cenkalti/backoff/v4 v4.2.1 // indirect github.com/containerd/containerd v1.7.3 // indirect github.com/cpuguy83/dockercfg v0.3.1 // indirect + github.com/davecgh/go-spew v1.1.1 // indirect github.com/docker/distribution v2.8.2+incompatible // indirect github.com/docker/docker v24.0.6+incompatible // indirect - github.com/docker/go-connections v0.4.0 // indirect github.com/docker/go-units v0.5.0 // indirect + github.com/eapache/go-resiliency v1.4.0 // indirect + github.com/eapache/go-xerial-snappy v0.0.0-20230731223053-c322873962e3 // indirect + github.com/eapache/queue v1.1.0 // indirect github.com/gogo/protobuf v1.3.2 // indirect github.com/golang/protobuf v1.5.3 // indirect + github.com/golang/snappy v0.0.4 // indirect github.com/google/uuid v1.3.1 // indirect - github.com/klauspost/compress v1.16.0 // indirect + github.com/hashicorp/errwrap v1.1.0 // indirect + github.com/hashicorp/go-multierror v1.1.1 // indirect + github.com/hashicorp/go-uuid v1.0.3 // indirect + github.com/jcmturner/aescts/v2 v2.0.0 // indirect + github.com/jcmturner/dnsutils/v2 v2.0.0 // indirect + github.com/jcmturner/gofork v1.7.6 // indirect + github.com/jcmturner/gokrb5/v8 v8.4.4 // indirect + github.com/jcmturner/rpc/v2 v2.0.3 // indirect + github.com/klauspost/compress v1.16.7 // indirect github.com/magiconair/properties v1.8.7 // indirect github.com/moby/patternmatcher v0.5.0 // indirect github.com/moby/sys/sequential v0.5.0 // indirect @@ -27,11 +43,14 @@ require ( github.com/opencontainers/go-digest v1.0.0 // indirect github.com/opencontainers/image-spec v1.1.0-rc4 // indirect github.com/opencontainers/runc v1.1.5 // indirect + github.com/pierrec/lz4/v4 v4.1.18 // indirect github.com/pkg/errors v0.9.1 // indirect + github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 // indirect github.com/sirupsen/logrus v1.9.0 // indirect + golang.org/x/crypto v0.12.0 // indirect golang.org/x/exp v0.0.0-20230510235704-dd950f8aeaea // indirect golang.org/x/mod v0.9.0 // indirect - golang.org/x/net v0.9.0 // indirect + golang.org/x/net v0.14.0 // indirect golang.org/x/sys v0.11.0 // indirect golang.org/x/tools v0.7.0 // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20230525234030-28d5490b6b19 // indirect diff --git a/modules/kafka/go.sum b/modules/kafka/go.sum index 8b6d587397..95074f6754 100644 --- a/modules/kafka/go.sum +++ b/modules/kafka/go.sum @@ -4,6 +4,8 @@ github.com/AdaLogics/go-fuzz-headers v0.0.0-20230106234847-43070de90fa1 h1:EKPd1 github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1 h1:UQHMgLO+TxOElx5B5HZ4hJQsoJ/PvUvKRhJHDQXO8P8= github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1/go.mod h1:xomTg63KZ2rFqZQzSB4Vz2SUXa1BpHTVz9L5PTmPC4E= github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU= +github.com/IBM/sarama v1.41.1 h1:B4/TdHce/8Ipza+qrLIeNJ9D1AOxZVp/3uDv6H/dp2M= +github.com/IBM/sarama v1.41.1/go.mod h1:JFCPURVskaipJdKRFkiE/OZqQHw7jqliaJmRwXCmSSw= github.com/Microsoft/go-winio v0.6.1 h1:9/kr64B9VUZrLm5YYwbGtUJnMgqWVOdUAXu6Migciow= github.com/Microsoft/go-winio v0.6.1/go.mod h1:LRdKpFKfdobln8UmuiYcKPot9D2v6svN5+sAH+4kjUM= github.com/Microsoft/hcsshim v0.10.0-rc.8 h1:YSZVvlIIDD1UxQpJp0h+dnpLUw+TrY0cx8obKsp3bek= @@ -33,6 +35,13 @@ github.com/docker/go-connections v0.4.0/go.mod h1:Gbd7IOopHjR8Iph03tsViu4nIes5Xh github.com/docker/go-units v0.4.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk= github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4= github.com/docker/go-units v0.5.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk= +github.com/eapache/go-resiliency v1.4.0 h1:3OK9bWpPk5q6pbFAaYSEwD9CLUSHG8bnZuqX2yMt3B0= +github.com/eapache/go-resiliency v1.4.0/go.mod h1:5yPzW0MIvSe0JDsv0v+DvcjEv2FyD6iZYSs1ZI+iQho= +github.com/eapache/go-xerial-snappy v0.0.0-20230731223053-c322873962e3 h1:Oy0F4ALJ04o5Qqpdz8XLIpNA3WM/iSIXqxtqo7UGVws= +github.com/eapache/go-xerial-snappy v0.0.0-20230731223053-c322873962e3/go.mod h1:YvSRo5mw33fLEx1+DlK6L2VV43tJt5Eyel9n9XBcR+0= +github.com/eapache/queue v1.1.0 h1:YOEu7KNc61ntiQlcEeUIoDTJ2o8mQznoNvUhiigpIqc= +github.com/eapache/queue v1.1.0/go.mod h1:6eCeP0CKFpHLu8blIFXhExK/dRa7WDZfr6jVFPTqq+I= +github.com/fortytw2/leaktest v1.3.0 h1:u8491cBMTQ8ft8aeV+adlcytMZylmA5nnwwkRZjI8vw= github.com/frankban/quicktest v1.11.3/go.mod h1:wRf/ReqHper53s+kmmSZizM8NamnL3IM0I9ntUbOk+k= github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA= github.com/godbus/dbus/v5 v5.0.6/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA= @@ -41,15 +50,39 @@ github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69 github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk= github.com/golang/protobuf v1.5.3 h1:KhyjKVUg7Usr/dYsdSqoFveMYd5ko72D+zANwlG1mmg= github.com/golang/protobuf v1.5.3/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY= +github.com/golang/snappy v0.0.4 h1:yAGX7huGHXlcLOEtBnF4w7FQwA26wojNCwOYAEhLjQM= +github.com/golang/snappy v0.0.4/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= github.com/google/go-cmp v0.5.4/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38= github.com/google/uuid v1.3.1 h1:KjJaJ9iWZ3jOFZIf1Lqf4laDRCasjl0BCmnEGxkdLb4= github.com/google/uuid v1.3.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/gorilla/securecookie v1.1.1/go.mod h1:ra0sb63/xPlUeL+yeDciTfxMRAA+MP+HVt/4epWDjd4= +github.com/gorilla/sessions v1.2.1/go.mod h1:dk2InVEVJ0sfLlnXv9EAgkf6ecYs/i80K/zI+bUmuGM= +github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4= +github.com/hashicorp/errwrap v1.1.0 h1:OxrOeh75EUXMY8TBjag2fzXGZ40LB6IKw45YeGUDY2I= +github.com/hashicorp/errwrap v1.1.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4= +github.com/hashicorp/go-multierror v1.1.1 h1:H5DkEtf6CXdFp0N0Em5UCwQpXMWke8IA0+lD48awMYo= +github.com/hashicorp/go-multierror v1.1.1/go.mod h1:iw975J/qwKPdAO1clOe2L8331t/9/fmwbPZ6JB6eMoM= +github.com/hashicorp/go-uuid v1.0.2/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro= +github.com/hashicorp/go-uuid v1.0.3 h1:2gKiV6YVmrJ1i2CKKa9obLvRieoRGviZFL26PcT/Co8= +github.com/hashicorp/go-uuid v1.0.3/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro= +github.com/jcmturner/aescts/v2 v2.0.0 h1:9YKLH6ey7H4eDBXW8khjYslgyqG2xZikXP0EQFKrle8= +github.com/jcmturner/aescts/v2 v2.0.0/go.mod h1:AiaICIRyfYg35RUkr8yESTqvSy7csK90qZ5xfvvsoNs= +github.com/jcmturner/dnsutils/v2 v2.0.0 h1:lltnkeZGL0wILNvrNiVCR6Ro5PGU/SeBvVO/8c/iPbo= +github.com/jcmturner/dnsutils/v2 v2.0.0/go.mod h1:b0TnjGOvI/n42bZa+hmXL+kFJZsFT7G4t3HTlQ184QM= +github.com/jcmturner/gofork v1.7.6 h1:QH0l3hzAU1tfT3rZCnW5zXl+orbkNMMRGJfdJjHVETg= +github.com/jcmturner/gofork v1.7.6/go.mod h1:1622LH6i/EZqLloHfE7IeZ0uEJwMSUyQ/nDd82IeqRo= +github.com/jcmturner/goidentity/v6 v6.0.1 h1:VKnZd2oEIMorCTsFBnJWbExfNN7yZr3EhJAxwOkZg6o= +github.com/jcmturner/goidentity/v6 v6.0.1/go.mod h1:X1YW3bgtvwAXju7V3LCIMpY0Gbxyjn/mY9zx4tFonSg= +github.com/jcmturner/gokrb5/v8 v8.4.4 h1:x1Sv4HaTpepFkXbt2IkL29DXRf8sOfZXo8eRKh687T8= +github.com/jcmturner/gokrb5/v8 v8.4.4/go.mod h1:1btQEpgT6k+unzCwX1KdWMEwPPkkgBtP+F6aCACiMrs= +github.com/jcmturner/rpc/v2 v2.0.3 h1:7FXXj8Ti1IaVFpSAziCZWNzbNuZmnvw/i6CqLNdWfZY= +github.com/jcmturner/rpc/v2 v2.0.3/go.mod h1:VUJYCIDm3PVOEHw8sgt091/20OJjskO/YJki3ELg/Hc= github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8= github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck= -github.com/klauspost/compress v1.16.0 h1:iULayQNOReoYUe+1qtKOqw9CwJv3aNQu8ivo7lw1HU4= -github.com/klauspost/compress v1.16.0/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE= +github.com/klauspost/compress v1.16.7 h1:2mk3MPGNzKyxErAw8YaohYh69+pa4sIQSC0fPGCFR9I= +github.com/klauspost/compress v1.16.7/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE= github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI= github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= @@ -73,10 +106,14 @@ github.com/opencontainers/runc v1.1.5 h1:L44KXEpKmfWDcS02aeGm8QNTFXTo2D+8MYGDIJ/ github.com/opencontainers/runc v1.1.5/go.mod h1:1J5XiS+vdZ3wCyZybsuxXZWGrgSr8fFJHLXuG2PsnNg= github.com/opencontainers/runtime-spec v1.0.3-0.20210326190908-1c3f411f0417/go.mod h1:jwyrGlmzljRJv/Fgzds9SsS/C5hL+LL3ko9hs6T5lQ0= github.com/opencontainers/selinux v1.10.0/go.mod h1:2i0OySw99QjzBBQByd1Gr9gSjvuho1lHsJxIJ3gGbJI= +github.com/pierrec/lz4/v4 v4.1.18 h1:xaKrnTkyoqfh1YItXl56+6KJNVYWlEEPuAQW9xsplYQ= +github.com/pierrec/lz4/v4 v4.1.18/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4= github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 h1:N/ElC8H3+5XpJzTSTfLsJV/mx9Q9g7kxmchpfZyxgzM= +github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4= github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM= github.com/seccomp/libseccomp-golang v0.9.2-0.20220502022130-f33da4d89646/go.mod h1:JA8cRccbGaA1s33RQf7Y1+q9gHmZX1yB/z9WDN1C6fg= github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeVRXNmyL/1OwPU0+IJeTBvfc= @@ -84,8 +121,14 @@ github.com/sirupsen/logrus v1.8.1/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic github.com/sirupsen/logrus v1.9.0 h1:trlNQbNUG3OdDrDil03MCb1H2o9nJ1x4/5LYw7byDE0= github.com/sirupsen/logrus v1.9.0/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw= +github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo= github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs= +github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4= github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= +github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4= github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk= github.com/syndtr/gocapability v0.0.0-20200815063812-42c35b437635/go.mod h1:hkRG7XYTFWNJGYcbNJQlaLq0fg1yr4J4t/NcTQtrfww= github.com/urfave/cli v1.22.1/go.mod h1:Gos4lmkARVdJ6EkW0WaNv/tZAAMe9V7XWyB60NtXRu0= @@ -93,26 +136,38 @@ github.com/vishvananda/netlink v1.1.0/go.mod h1:cTgwzPIzzgDAYoQrMm0EdrjRUBkTqKYp github.com/vishvananda/netns v0.0.0-20191106174202-0a2b9b5464df/go.mod h1:JP3t17pCcGlemwknint6hfoeCVQrEMVwxRLRjXpq+BU= github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= +github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY= golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= +golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc= +golang.org/x/crypto v0.6.0/go.mod h1:OFC/31mSvZgRz0V1QTNCzfAI1aIRzbiufJtkMIlEp58= +golang.org/x/crypto v0.12.0 h1:tFM/ta59kqch6LlvYnPa0yx5a83cL2nHflFhYKvv9Yk= +golang.org/x/crypto v0.12.0/go.mod h1:NF0Gs7EO5K4qLn+Ylc+fih8BSTeIjAP05siRnAh98yw= golang.org/x/exp v0.0.0-20230510235704-dd950f8aeaea h1:vLCWI/yYrdEHyN2JzIzPO3aaQJHQdp89IZBA/+azVC4= golang.org/x/exp v0.0.0-20230510235704-dd950f8aeaea/go.mod h1:V1LtkGg67GoY2N1AnLN78QLrzxkLyJw7RJb1gzOOz9w= golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= +golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4= golang.org/x/mod v0.9.0 h1:KENHtAZL2y3NLMYZeHY9DW8HW8V+kQyJsY/V9JlKvCs= golang.org/x/mod v0.9.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs= golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.0.0-20200114155413-6afb5195e5aa/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU= golang.org/x/net v0.0.0-20201224014010-6772e930b67b/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg= -golang.org/x/net v0.9.0 h1:aWJ/m6xSmxWBx+V0XRHTlrYrPG56jKsLdTFmsSsCzOM= -golang.org/x/net v0.9.0/go.mod h1:d48xBJpPfHeWQsugry2m+kC02ZBRGRgulfHnEXEuWns= +golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg= +golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c= +golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs= +golang.org/x/net v0.7.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs= +golang.org/x/net v0.14.0 h1:BONx9s002vGdD9umnlX1Po8vOZmrgH34qlHcD1MfK14= +golang.org/x/net v0.14.0/go.mod h1:PpSgVXXLK0OxS0F31C1/tv6XNguvCrnXIDrFMspZIUI= golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= -golang.org/x/sync v0.1.0 h1:wsuoTGHzEhffawBOhz5CYhcrV4IdKZbEyZjBMuTp12o= +golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.3.0 h1:ftCYgMx6zT/asHUrPw8BLLscYtGznsLAnjq5RH9P66E= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20190606203320-7fc4e5ec1444/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= @@ -121,22 +176,31 @@ golang.org/x/sys v0.0.0-20191115151921-52ab43148777/go.mod h1:h1NjWce9XRLGQEsW7w golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20210124154548-22da62e12c0c/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20210616094352-59db8d763f22/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20210906170528-6f6e22806c34/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20211025201205-69cdffdb9359/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20211116061358-0a5406a5449c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.11.0 h1:eG7RXZHdqOJ1i+0lgLgCpSXAp6M3LYlAo6osgSi0xOM= golang.org/x/sys v0.11.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= +golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= +golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= -golang.org/x/text v0.9.0 h1:2sjJmO8cDvYveuX97RDLsxlyUxLl+GHoLxBiRdHllBE= +golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ= +golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8= +golang.org/x/text v0.12.0 h1:k+n5B8goJNdU7hSvEtMUz3d1Q6D/XW4COJSJR6fN0mc= golang.org/x/time v0.0.0-20220210224613-90d013bbcef8 h1:vVKdlvoWBphwdxWKrFZEuM0kGgGLxUOYcY4U/2Vjg44= golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE= golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= +golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc= golang.org/x/tools v0.7.0 h1:W4OVu8VVOaIO0yzWMNdepAulS7YfoS3Zabrm8DOXXU4= golang.org/x/tools v0.7.0/go.mod h1:4pg6aUX35JBAogB10C9AtvVL+qowtN4pT3CGSQex14s= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= diff --git a/modules/kafka/kafka.go b/modules/kafka/kafka.go index 9eeaf9c657..84407888bf 100644 --- a/modules/kafka/kafka.go +++ b/modules/kafka/kafka.go @@ -2,8 +2,28 @@ package kafka import ( "context" + "fmt" + "math" + + "github.com/docker/go-connections/nat" "github.com/testcontainers/testcontainers-go" + "github.com/testcontainers/testcontainers-go/wait" +) + +const publicPort = nat.Port("9093/tcp") +const ( + starterScript = "/usr/sbin/testcontainers_start.sh" + + starterScriptContent = `#!/bin/bash +source /etc/confluent/docker/bash-config +export KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://%s:%d,BROKER://%s:9092 +echo Starting Kafka KRaft mode +sed -i '/KAFKA_ZOOKEEPER_CONNECT/d' /etc/confluent/docker/configure +echo 'kafka-storage format --ignore-formatted -t "$(kafka-storage random-uuid)" -c /etc/kafka/kafka.properties' >> /etc/confluent/docker/configure +echo '' > /etc/confluent/docker/ensure +/etc/confluent/docker/configure +/etc/confluent/docker/launch` ) // KafkaContainer represents the Kafka container type used in the module @@ -14,7 +34,53 @@ type KafkaContainer struct { // RunContainer creates an instance of the Kafka container type func RunContainer(ctx context.Context, opts ...testcontainers.ContainerCustomizer) (*KafkaContainer, error) { req := testcontainers.ContainerRequest{ - Image: "confluentinc/cp-kafka:7.3.3", + Image: "confluentinc/cp-kafka:7.3.3", + ExposedPorts: []string{string(publicPort)}, + Env: map[string]string{ + "KAFKA_LISTENERS": "PLAINTEXT://0.0.0.0:9093,BROKER://0.0.0.0:9092,CONTROLLER://0.0.0.0:9094", + "KAFKA_LISTENER_SECURITY_PROTOCOL_MAP": "BROKER:PLAINTEXT,PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT", + "KAFKA_INTER_BROKER_LISTENER_NAME": "BROKER", + "KAFKA_BROKER_ID": "1", + "KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR": "1", + "KAFKA_OFFSETS_TOPIC_NUM_PARTITIONS": "1", + "KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR": "1", + "KAFKA_TRANSACTION_STATE_LOG_MIN_ISR": "1", + "KAFKA_LOG_FLUSH_INTERVAL_MESSAGES": fmt.Sprintf("%d", math.MaxInt64), + "KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS": "0", + "KAFKA_NODE_ID": "1", + "KAFKA_PROCESS_ROLES": "broker,controller", + "KAFKA_CONTROLLER_QUORUM_VOTERS": "1@localhost:9094", + "KAFKA_CONTROLLER_LISTENER_NAMES": "CONTROLLER", + }, + Entrypoint: []string{"sh"}, + // this CMD will wait for the starter script to be copied into the container and then execute it + Cmd: []string{"-c", "while [ ! -f " + starterScript + " ]; do sleep 0.1; done; bash " + starterScript}, + LifecycleHooks: []testcontainers.ContainerLifecycleHooks{ + { + PostStarts: []testcontainers.ContainerHook{ + // 1. copy the starter script into the container + func(ctx context.Context, c testcontainers.Container) error { + host, err := c.Host(ctx) + if err != nil { + return err + } + + port, err := c.MappedPort(ctx, publicPort) + if err != nil { + return err + } + + scriptContent := fmt.Sprintf(starterScriptContent, host, port.Int(), host) + + return c.CopyToContainer(ctx, []byte(scriptContent), starterScript, 0o755) + }, + // 2. wait for the Kafka server to be ready + func(ctx context.Context, c testcontainers.Container) error { + return wait.ForLog("Kafka Server started").WaitUntilReady(ctx, c) + }, + }, + }, + }, } genericContainerReq := testcontainers.GenericContainerRequest{ @@ -33,3 +99,17 @@ func RunContainer(ctx context.Context, opts ...testcontainers.ContainerCustomize return &KafkaContainer{Container: container}, nil } + +// Brokers retrieves the broker connection strings from Kafka +func (kc *KafkaContainer) Brokers(ctx context.Context) ([]string, error) { + host, err := kc.Host(ctx) + if err != nil { + return nil, err + } + + port, err := kc.MappedPort(ctx, publicPort) + if err != nil { + return nil, err + } + return []string{fmt.Sprintf("%s:%d", host, port.Int())}, nil +} diff --git a/modules/kafka/kafka_test.go b/modules/kafka/kafka_test.go index 61fbec9783..71d6b42f87 100644 --- a/modules/kafka/kafka_test.go +++ b/modules/kafka/kafka_test.go @@ -2,23 +2,77 @@ package kafka import ( "context" + "strings" "testing" + + "github.com/IBM/sarama" + "github.com/testcontainers/testcontainers-go" ) func TestKafka(t *testing.T) { + topic := "some-topic" + ctx := context.Background() - container, err := RunContainer(ctx) + kafkaContainer, err := RunContainer(ctx, testcontainers.WithImage("confluentinc/cp-kafka:7.3.3")) if err != nil { t.Fatal(err) } // Clean up the container after the test is complete t.Cleanup(func() { - if err := container.Terminate(ctx); err != nil { + if err := kafkaContainer.Terminate(ctx); err != nil { t.Fatalf("failed to terminate container: %s", err) } }) + brokers, err := kafkaContainer.Brokers(ctx) + if err != nil { + t.Fatal(err) + } + + config := sarama.NewConfig() + client, err := sarama.NewConsumerGroup(brokers, "groupName", config) + if err != nil { + t.Fatal(err) + } + + consumer, ready, done, cancel := NewTestKafkaConsumer(t) + go func() { + if err := client.Consume(context.Background(), []string{topic}, consumer); err != nil { + cancel() + } + }() + + // wait for the consumer to be ready + <-ready + // perform assertions + + // set config to true because successfully delivered messages will be returned on the Successes channel + config.Producer.Return.Successes = true + + producer, err := sarama.NewSyncProducer(brokers, config) + if err != nil { + cancel() + t.Fatal(err) + } + + if _, _, err := producer.SendMessage(&sarama.ProducerMessage{ + Topic: topic, + Key: sarama.StringEncoder("keys"), + Value: sarama.StringEncoder("value"), + }); err != nil { + cancel() + t.Fatal(err) + } + + <-done + + if !strings.EqualFold(string(consumer.message.Key), "key") { + t.Fatalf("expected key to be %s, got %s", "key", string(consumer.message.Key)) + } + if !strings.EqualFold(string(consumer.message.Value), "value") { + t.Fatalf("expected value to be %s, got %s", "value", string(consumer.message.Value)) + } } diff --git a/modules/kafka/testing.go b/modules/kafka/testing.go new file mode 100644 index 0000000000..48d367e5ed --- /dev/null +++ b/modules/kafka/testing.go @@ -0,0 +1,55 @@ +package kafka + +import ( + "testing" + + "github.com/IBM/sarama" +) + +// TestKafkaConsumer is a test consumer for Kafka +type TestKafkaConsumer struct { + t *testing.T + ready chan bool + done chan bool + cancel chan bool + message *sarama.ConsumerMessage +} + +func NewTestKafkaConsumer(t *testing.T) (consumer *TestKafkaConsumer, ready <-chan bool, done <-chan bool, cancel func()) { + kc := &TestKafkaConsumer{ + t: t, + ready: make(chan bool, 1), + done: make(chan bool, 1), + cancel: make(chan bool, 1), + } + return kc, kc.ready, kc.done, func() { + kc.cancel <- true + } +} + +func (k *TestKafkaConsumer) Setup(_ sarama.ConsumerGroupSession) error { + return nil +} + +func (k *TestKafkaConsumer) Cleanup(_ sarama.ConsumerGroupSession) error { + return nil +} + +// ConsumeClaim is called by the Kafka client library when a message is received +func (k *TestKafkaConsumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error { + k.ready <- true + for { + select { + case message := <-claim.Messages(): + k.message = message + session.MarkMessage(message, "") + k.done <- true + + case <-k.cancel: + return nil + + case <-session.Context().Done(): + return nil + } + } +} From 3450dc83f9910b11c66ee753636f170169b31d55 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Manuel=20de=20la=20Pe=C3=B1a?= Date: Tue, 12 Sep 2023 04:14:36 +0200 Subject: [PATCH 03/16] docs: document methods and options --- docs/modules/kafka.md | 24 ++++++++++++++++++++++++ modules/kafka/kafka.go | 8 +++++++- modules/kafka/kafka_test.go | 2 ++ 3 files changed, 33 insertions(+), 1 deletion(-) diff --git a/docs/modules/kafka.md b/docs/modules/kafka.md index c5085163fe..f32ded6b66 100644 --- a/docs/modules/kafka.md +++ b/docs/modules/kafka.md @@ -40,6 +40,22 @@ When starting the Kafka container, you can pass options in a variadic way to con If you need to set a different Kafka Docker image, you can use `testcontainers.WithImage` with a valid Docker image for Kafka. E.g. `testcontainers.WithImage("confluentinc/cp-kafka:7.3.3")`. +#### Init script + +The Kafka container will be started using a custom shell script: + + +[Init script](../../modules/kafka/kafka.go) inside_block:starterScript + + +#### Environment variables + +The environment variables that are already set by default are: + + +[Environment variables](../../modules/kafka/kafka.go) inside_block:envVars + + #### Wait Strategies If you need to set a different wait strategy for Kafka, you can use `testcontainers.WithWaitStrategy` with a valid wait strategy @@ -63,3 +79,11 @@ Please read the [Create containers: Advanced Settings](../features/creating_cont ### Container Methods The Kafka container exposes the following methods: + +#### Brokers + +The `Brokers(ctx)` method returns the Kafka brokers as a string slice, containing the host and the random port defined by Kafka's public port (`9093/tcp`). + + +[Get Kafka brokers](../../modules/kafka/kafka_test.go) inside_block:getBrokers + diff --git a/modules/kafka/kafka.go b/modules/kafka/kafka.go index 84407888bf..e55a387cf9 100644 --- a/modules/kafka/kafka.go +++ b/modules/kafka/kafka.go @@ -15,6 +15,7 @@ const publicPort = nat.Port("9093/tcp") const ( starterScript = "/usr/sbin/testcontainers_start.sh" + // starterScript { starterScriptContent = `#!/bin/bash source /etc/confluent/docker/bash-config export KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://%s:%d,BROKER://%s:9092 @@ -24,6 +25,7 @@ echo 'kafka-storage format --ignore-formatted -t "$(kafka-storage random-uuid)" echo '' > /etc/confluent/docker/ensure /etc/confluent/docker/configure /etc/confluent/docker/launch` + // } ) // KafkaContainer represents the Kafka container type used in the module @@ -37,6 +39,7 @@ func RunContainer(ctx context.Context, opts ...testcontainers.ContainerCustomize Image: "confluentinc/cp-kafka:7.3.3", ExposedPorts: []string{string(publicPort)}, Env: map[string]string{ + // envVars { "KAFKA_LISTENERS": "PLAINTEXT://0.0.0.0:9093,BROKER://0.0.0.0:9092,CONTROLLER://0.0.0.0:9094", "KAFKA_LISTENER_SECURITY_PROTOCOL_MAP": "BROKER:PLAINTEXT,PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT", "KAFKA_INTER_BROKER_LISTENER_NAME": "BROKER", @@ -51,6 +54,7 @@ func RunContainer(ctx context.Context, opts ...testcontainers.ContainerCustomize "KAFKA_PROCESS_ROLES": "broker,controller", "KAFKA_CONTROLLER_QUORUM_VOTERS": "1@localhost:9094", "KAFKA_CONTROLLER_LISTENER_NAMES": "CONTROLLER", + // } }, Entrypoint: []string{"sh"}, // this CMD will wait for the starter script to be copied into the container and then execute it @@ -100,7 +104,8 @@ func RunContainer(ctx context.Context, opts ...testcontainers.ContainerCustomize return &KafkaContainer{Container: container}, nil } -// Brokers retrieves the broker connection strings from Kafka +// Brokers retrieves the broker connection strings from Kafka with only one entry, +// defined by the exposed public port. func (kc *KafkaContainer) Brokers(ctx context.Context) ([]string, error) { host, err := kc.Host(ctx) if err != nil { @@ -111,5 +116,6 @@ func (kc *KafkaContainer) Brokers(ctx context.Context) ([]string, error) { if err != nil { return nil, err } + return []string{fmt.Sprintf("%s:%d", host, port.Int())}, nil } diff --git a/modules/kafka/kafka_test.go b/modules/kafka/kafka_test.go index 71d6b42f87..daba44424d 100644 --- a/modules/kafka/kafka_test.go +++ b/modules/kafka/kafka_test.go @@ -26,7 +26,9 @@ func TestKafka(t *testing.T) { } }) + // getBrokers { brokers, err := kafkaContainer.Brokers(ctx) + // } if err != nil { t.Fatal(err) } From 0e187babe4d0ee6b4d139ae350c99b6dd70fbb74 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Manuel=20de=20la=20Pe=C3=B1a?= Date: Tue, 12 Sep 2023 04:41:03 +0200 Subject: [PATCH 04/16] fix: golangci-lint --- modules/kafka/kafka_test.go | 1 + 1 file changed, 1 insertion(+) diff --git a/modules/kafka/kafka_test.go b/modules/kafka/kafka_test.go index daba44424d..1e1458d14d 100644 --- a/modules/kafka/kafka_test.go +++ b/modules/kafka/kafka_test.go @@ -6,6 +6,7 @@ import ( "testing" "github.com/IBM/sarama" + "github.com/testcontainers/testcontainers-go" ) From df62732fe8af2712e0ea11a55d50095fd98fa959 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Manuel=20de=20la=20Pe=C3=B1a?= Date: Tue, 12 Sep 2023 04:51:40 +0200 Subject: [PATCH 05/16] fix: wrong copy&paste --- modules/kafka/kafka_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/modules/kafka/kafka_test.go b/modules/kafka/kafka_test.go index 1e1458d14d..160536eddf 100644 --- a/modules/kafka/kafka_test.go +++ b/modules/kafka/kafka_test.go @@ -63,7 +63,7 @@ func TestKafka(t *testing.T) { if _, _, err := producer.SendMessage(&sarama.ProducerMessage{ Topic: topic, - Key: sarama.StringEncoder("keys"), + Key: sarama.StringEncoder("key"), Value: sarama.StringEncoder("value"), }); err != nil { cancel() From 0ee48838ebfbc6c0a5e1bc76bd93fd9f7fa269bf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Manuel=20de=20la=20Pe=C3=B1a?= Date: Tue, 12 Sep 2023 10:13:33 +0200 Subject: [PATCH 06/16] docs: link to KRaft docs --- docs/modules/kafka.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/docs/modules/kafka.md b/docs/modules/kafka.md index f32ded6b66..147e9af6d6 100644 --- a/docs/modules/kafka.md +++ b/docs/modules/kafka.md @@ -1,10 +1,10 @@ -# Kafka +# Kafka (KRaft) Not available until the next release of testcontainers-go :material-tag: main ## Introduction -The Testcontainers module for Kafka. +The Testcontainers module for KRaft: [Apache Kafka Without ZooKeeper](https://developer.confluent.io/learn/kraft). ## Adding this module to your project dependencies From 5c012e5b6b60a886eae97596a57955b934302105 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Manuel=20de=20la=20Pe=C3=B1a?= Date: Tue, 12 Sep 2023 10:13:46 +0200 Subject: [PATCH 07/16] chore: better wait for KRaft log --- modules/kafka/kafka.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/modules/kafka/kafka.go b/modules/kafka/kafka.go index e55a387cf9..a0f20e4364 100644 --- a/modules/kafka/kafka.go +++ b/modules/kafka/kafka.go @@ -80,7 +80,7 @@ func RunContainer(ctx context.Context, opts ...testcontainers.ContainerCustomize }, // 2. wait for the Kafka server to be ready func(ctx context.Context, c testcontainers.Container) error { - return wait.ForLog("Kafka Server started").WaitUntilReady(ctx, c) + return wait.ForLog(".*Transitioning from RECOVERY to RUNNING.*").AsRegexp().WaitUntilReady(ctx, c) }, }, }, From f1c1e9e0a5cd7dc60302fb50d3e811c6daebfc23 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Manuel=20de=20la=20Pe=C3=B1a?= Date: Tue, 12 Sep 2023 10:44:39 +0200 Subject: [PATCH 08/16] chore: support for set cluster ID --- modules/kafka/examples_test.go | 7 ++++++- modules/kafka/kafka.go | 11 ++++++++++- modules/kafka/kafka_test.go | 6 +++++- 3 files changed, 21 insertions(+), 3 deletions(-) diff --git a/modules/kafka/examples_test.go b/modules/kafka/examples_test.go index 39e47e404c..c5a23b289a 100644 --- a/modules/kafka/examples_test.go +++ b/modules/kafka/examples_test.go @@ -12,7 +12,10 @@ func ExampleRunContainer() { // runKafkaContainer { ctx := context.Background() - kafkaContainer, err := kafka.RunContainer(ctx, testcontainers.WithImage("confluentinc/cp-kafka:7.3.3")) + kafkaContainer, err := kafka.RunContainer(ctx, + kafka.WithClusterID("test-cluster"), + testcontainers.WithImage("confluentinc/cp-kafka:7.3.3"), + ) if err != nil { panic(err) } @@ -30,8 +33,10 @@ func ExampleRunContainer() { panic(err) } + fmt.Println(kafkaContainer.ClusterID) fmt.Println(state.Running) // Output: + // test-cluster // true } diff --git a/modules/kafka/kafka.go b/modules/kafka/kafka.go index a0f20e4364..7616b160dc 100644 --- a/modules/kafka/kafka.go +++ b/modules/kafka/kafka.go @@ -31,6 +31,7 @@ echo '' > /etc/confluent/docker/ensure // KafkaContainer represents the Kafka container type used in the module type KafkaContainer struct { testcontainers.Container + ClusterID string } // RunContainer creates an instance of the Kafka container type @@ -96,12 +97,20 @@ func RunContainer(ctx context.Context, opts ...testcontainers.ContainerCustomize opt.Customize(&genericContainerReq) } + clusterID := genericContainerReq.Env["CLUSTER_ID"] + container, err := testcontainers.GenericContainer(ctx, genericContainerReq) if err != nil { return nil, err } - return &KafkaContainer{Container: container}, nil + return &KafkaContainer{Container: container, ClusterID: clusterID}, nil +} + +func WithClusterID(clusterID string) testcontainers.CustomizeRequestOption { + return func(req *testcontainers.GenericContainerRequest) { + req.Env["CLUSTER_ID"] = clusterID + } } // Brokers retrieves the broker connection strings from Kafka with only one entry, diff --git a/modules/kafka/kafka_test.go b/modules/kafka/kafka_test.go index 160536eddf..60d2b59434 100644 --- a/modules/kafka/kafka_test.go +++ b/modules/kafka/kafka_test.go @@ -15,7 +15,7 @@ func TestKafka(t *testing.T) { ctx := context.Background() - kafkaContainer, err := RunContainer(ctx, testcontainers.WithImage("confluentinc/cp-kafka:7.3.3")) + kafkaContainer, err := RunContainer(ctx, WithClusterID("kraftCluster"), testcontainers.WithImage("confluentinc/cp-kafka:7.3.3")) if err != nil { t.Fatal(err) } @@ -27,6 +27,10 @@ func TestKafka(t *testing.T) { } }) + if !strings.EqualFold(kafkaContainer.ClusterID, "kraftCluster") { + t.Fatalf("expected clusterID to be %s, got %s", "kraftCluster", kafkaContainer.ClusterID) + } + // getBrokers { brokers, err := kafkaContainer.Brokers(ctx) // } From 984ea3410d78fce1f97be65e1d1cebbf968dab95 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Manuel=20de=20la=20Pe=C3=B1a?= Date: Tue, 12 Sep 2023 11:59:53 +0200 Subject: [PATCH 09/16] chore: configure controller quorum voters based on the networks --- modules/kafka/kafka.go | 25 +++++++++++++++- modules/kafka/kafka_test.go | 57 +++++++++++++++++++++++++++++++++++++ 2 files changed, 81 insertions(+), 1 deletion(-) diff --git a/modules/kafka/kafka.go b/modules/kafka/kafka.go index 7616b160dc..8b27c48922 100644 --- a/modules/kafka/kafka.go +++ b/modules/kafka/kafka.go @@ -53,7 +53,6 @@ func RunContainer(ctx context.Context, opts ...testcontainers.ContainerCustomize "KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS": "0", "KAFKA_NODE_ID": "1", "KAFKA_PROCESS_ROLES": "broker,controller", - "KAFKA_CONTROLLER_QUORUM_VOTERS": "1@localhost:9094", "KAFKA_CONTROLLER_LISTENER_NAMES": "CONTROLLER", // } }, @@ -99,6 +98,8 @@ func RunContainer(ctx context.Context, opts ...testcontainers.ContainerCustomize clusterID := genericContainerReq.Env["CLUSTER_ID"] + configureControllerQuorumVoters(&genericContainerReq) + container, err := testcontainers.GenericContainer(ctx, genericContainerReq) if err != nil { return nil, err @@ -128,3 +129,25 @@ func (kc *KafkaContainer) Brokers(ctx context.Context) ([]string, error) { return []string{fmt.Sprintf("%s:%d", host, port.Int())}, nil } + +// configureControllerQuorumVoters sets the quorum voters for the controller. For that, it will +// check if there are any network aliases defined for the container and use the first alias in the +// first network. Else, it will use localhost. +func configureControllerQuorumVoters(req *testcontainers.GenericContainerRequest) { + if req.Env == nil { + req.Env = map[string]string{} + } + + if req.Env["KAFKA_CONTROLLER_QUORUM_VOTERS"] == "" { + host := "localhost" + if len(req.Networks) > 0 { + nw := req.Networks[0] + if len(req.NetworkAliases[nw]) > 0 { + host = req.NetworkAliases[nw][0] + } + } + + req.Env["KAFKA_CONTROLLER_QUORUM_VOTERS"] = fmt.Sprintf("1@%s:9094", host) + } + // } +} diff --git a/modules/kafka/kafka_test.go b/modules/kafka/kafka_test.go index 60d2b59434..2f03987429 100644 --- a/modules/kafka/kafka_test.go +++ b/modules/kafka/kafka_test.go @@ -83,3 +83,60 @@ func TestKafka(t *testing.T) { t.Fatalf("expected value to be %s, got %s", "value", string(consumer.message.Value)) } } + +func TestConfigureQuorumVoters(t *testing.T) { + tests := []struct { + name string + req *testcontainers.GenericContainerRequest + expectedVoters string + }{ + { + name: "voters on localhost", + req: &testcontainers.GenericContainerRequest{ + ContainerRequest: testcontainers.ContainerRequest{ + Env: map[string]string{}, + }, + }, + expectedVoters: "1@localhost:9094", + }, + { + name: "voters on first network alias of the first network", + req: &testcontainers.GenericContainerRequest{ + ContainerRequest: testcontainers.ContainerRequest{ + Env: map[string]string{}, + Networks: []string{"foo", "bar", "baaz"}, + NetworkAliases: map[string][]string{ + "foo": {"foo0", "foo1", "foo2", "foo3"}, + "bar": {"bar0", "bar1", "bar2", "bar3"}, + "baaz": {"baaz0", "baaz1", "baaz2", "baaz3"}, + }, + }, + }, + expectedVoters: "1@foo0:9094", + }, + { + name: "voters on localhost if alias but no networks", + req: &testcontainers.GenericContainerRequest{ + ContainerRequest: testcontainers.ContainerRequest{ + NetworkAliases: map[string][]string{ + "foo": {"foo0", "foo1", "foo2", "foo3"}, + "bar": {"bar0", "bar1", "bar2", "bar3"}, + "baaz": {"baaz0", "baaz1", "baaz2", "baaz3"}, + }, + }, + }, + expectedVoters: "1@localhost:9094", + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + configureControllerQuorumVoters(test.req) + + if test.req.Env["KAFKA_CONTROLLER_QUORUM_VOTERS"] != test.expectedVoters { + t.Fatalf("expected KAFKA_CONTROLLER_QUORUM_VOTERS to be %s, got %s", test.expectedVoters, test.req.Env["KAFKA_CONTROLLER_QUORUM_VOTERS"]) + } + }) + } + +} From c03bfcc2eb55464e2a539ee75964e3536f7c1b90 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Manuel=20de=20la=20Pe=C3=B1a?= Date: Tue, 12 Sep 2023 12:14:08 +0200 Subject: [PATCH 10/16] chore: validate KRaft version is above 7.0.0 --- modules/kafka/go.mod | 2 +- modules/kafka/go.sum | 4 +-- modules/kafka/kafka.go | 29 +++++++++++++++++++++ modules/kafka/kafka_test.go | 52 +++++++++++++++++++++++++++++++++++++ 4 files changed, 84 insertions(+), 3 deletions(-) diff --git a/modules/kafka/go.mod b/modules/kafka/go.mod index f61f85af56..cf6ae0296d 100644 --- a/modules/kafka/go.mod +++ b/modules/kafka/go.mod @@ -6,6 +6,7 @@ require ( github.com/IBM/sarama v1.41.1 github.com/docker/go-connections v0.4.0 github.com/testcontainers/testcontainers-go v0.23.0 + golang.org/x/mod v0.12.0 ) require ( @@ -49,7 +50,6 @@ require ( github.com/sirupsen/logrus v1.9.0 // indirect golang.org/x/crypto v0.12.0 // indirect golang.org/x/exp v0.0.0-20230510235704-dd950f8aeaea // indirect - golang.org/x/mod v0.9.0 // indirect golang.org/x/net v0.14.0 // indirect golang.org/x/sys v0.11.0 // indirect golang.org/x/tools v0.7.0 // indirect diff --git a/modules/kafka/go.sum b/modules/kafka/go.sum index 95074f6754..6fadfef92b 100644 --- a/modules/kafka/go.sum +++ b/modules/kafka/go.sum @@ -149,8 +149,8 @@ golang.org/x/exp v0.0.0-20230510235704-dd950f8aeaea/go.mod h1:V1LtkGg67GoY2N1AnL golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4= -golang.org/x/mod v0.9.0 h1:KENHtAZL2y3NLMYZeHY9DW8HW8V+kQyJsY/V9JlKvCs= -golang.org/x/mod v0.9.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs= +golang.org/x/mod v0.12.0 h1:rmsUpXtvNzj340zd98LZ4KntptpfRHwpFOHG188oHXc= +golang.org/x/mod v0.12.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs= golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= golang.org/x/net v0.0.0-20200114155413-6afb5195e5aa/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= diff --git a/modules/kafka/kafka.go b/modules/kafka/kafka.go index 8b27c48922..b743f9cc37 100644 --- a/modules/kafka/kafka.go +++ b/modules/kafka/kafka.go @@ -4,9 +4,12 @@ import ( "context" "fmt" "math" + "strings" "github.com/docker/go-connections/nat" + "golang.org/x/mod/semver" + "github.com/testcontainers/testcontainers-go" "github.com/testcontainers/testcontainers-go/wait" ) @@ -96,6 +99,11 @@ func RunContainer(ctx context.Context, opts ...testcontainers.ContainerCustomize opt.Customize(&genericContainerReq) } + err := validateKRaftVersion(genericContainerReq.Image) + if err != nil { + return nil, err + } + clusterID := genericContainerReq.Env["CLUSTER_ID"] configureControllerQuorumVoters(&genericContainerReq) @@ -151,3 +159,24 @@ func configureControllerQuorumVoters(req *testcontainers.GenericContainerRequest } // } } + +// validateKRaftVersion validates if the image version is compatible with KRaft mode, +// which is available since version 7.0.0. +func validateKRaftVersion(image string) error { + if image == "" { + return fmt.Errorf("image cannot be empty") + } + + version := image[strings.LastIndex(image, ":")+1:] + + // semver requires the version to start with a "v" + if !strings.HasPrefix(version, "v") { + version = fmt.Sprintf("v%s", version) + } + + if semver.Compare(version, "v7.0.0") < 0 { // version < v7.0.0 + return fmt.Errorf("version=%s. KRaft mode is only available since version 7.0.0", version) + } + + return nil +} diff --git a/modules/kafka/kafka_test.go b/modules/kafka/kafka_test.go index 2f03987429..5904fab255 100644 --- a/modules/kafka/kafka_test.go +++ b/modules/kafka/kafka_test.go @@ -84,6 +84,15 @@ func TestKafka(t *testing.T) { } } +func TestKafka_invalidVersion(t *testing.T) { + ctx := context.Background() + + _, err := RunContainer(ctx, WithClusterID("kraftCluster"), testcontainers.WithImage("confluentinc/cp-kafka:6.3.3")) + if err == nil { + t.Fatal(err) + } +} + func TestConfigureQuorumVoters(t *testing.T) { tests := []struct { name string @@ -140,3 +149,46 @@ func TestConfigureQuorumVoters(t *testing.T) { } } + +func TestVAlidateKRaftVersion(t *testing.T) { + tests := []struct { + name string + version string + wantErr bool + }{ + { + name: "valid version", + version: "7.3.3", + wantErr: false, + }, + { + name: "valid version: limit", + version: "7.0.0", + wantErr: false, + }, + { + name: "invalid version: low version", + version: "6.99.99", + wantErr: true, + }, + { + name: "invalid version: too low", + version: "5.0.0", + wantErr: true, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + err := validateKRaftVersion("test-kafka:" + test.version) + + if test.wantErr && err == nil { + t.Fatalf("expected error, got nil") + } + + if !test.wantErr && err != nil { + t.Fatalf("expected no error, got %s", err) + } + }) + } +} From 02cf8e00980a798e225d79dff341d7f8c7d7c1d6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Manuel=20de=20la=20Pe=C3=B1a?= Date: Tue, 12 Sep 2023 12:22:04 +0200 Subject: [PATCH 11/16] fix: lint --- modules/kafka/kafka.go | 1 - modules/kafka/kafka_test.go | 1 - 2 files changed, 2 deletions(-) diff --git a/modules/kafka/kafka.go b/modules/kafka/kafka.go index b743f9cc37..42813a467f 100644 --- a/modules/kafka/kafka.go +++ b/modules/kafka/kafka.go @@ -7,7 +7,6 @@ import ( "strings" "github.com/docker/go-connections/nat" - "golang.org/x/mod/semver" "github.com/testcontainers/testcontainers-go" diff --git a/modules/kafka/kafka_test.go b/modules/kafka/kafka_test.go index 5904fab255..a242eca77c 100644 --- a/modules/kafka/kafka_test.go +++ b/modules/kafka/kafka_test.go @@ -147,7 +147,6 @@ func TestConfigureQuorumVoters(t *testing.T) { } }) } - } func TestVAlidateKRaftVersion(t *testing.T) { From c96b03fdacffd65bab1de4389711b4ad0ea17441 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Manuel=20de=20la=20Pe=C3=B1a?= Date: Tue, 12 Sep 2023 12:31:22 +0200 Subject: [PATCH 12/16] chore: validate that the image namespace is the official one --- modules/kafka/kafka.go | 14 +++++++++++--- modules/kafka/kafka_test.go | 25 +++++++++++++++---------- 2 files changed, 26 insertions(+), 13 deletions(-) diff --git a/modules/kafka/kafka.go b/modules/kafka/kafka.go index 42813a467f..45fd2a5fe5 100644 --- a/modules/kafka/kafka.go +++ b/modules/kafka/kafka.go @@ -161,12 +161,20 @@ func configureControllerQuorumVoters(req *testcontainers.GenericContainerRequest // validateKRaftVersion validates if the image version is compatible with KRaft mode, // which is available since version 7.0.0. -func validateKRaftVersion(image string) error { - if image == "" { +func validateKRaftVersion(fqName string) error { + if fqName == "" { return fmt.Errorf("image cannot be empty") } - version := image[strings.LastIndex(image, ":")+1:] + image := fqName[:strings.LastIndex(fqName, ":")] + version := fqName[strings.LastIndex(fqName, ":")+1:] + + if !strings.EqualFold(image, "confluentinc/cp-kafka") { + // do not validate if the image is not the official one. + // not raising an error here, letting the image to start and + // eventually evaluate an error if it exists. + return nil + } // semver requires the version to start with a "v" if !strings.HasPrefix(version, "v") { diff --git a/modules/kafka/kafka_test.go b/modules/kafka/kafka_test.go index a242eca77c..4edc9095e7 100644 --- a/modules/kafka/kafka_test.go +++ b/modules/kafka/kafka_test.go @@ -152,34 +152,39 @@ func TestConfigureQuorumVoters(t *testing.T) { func TestVAlidateKRaftVersion(t *testing.T) { tests := []struct { name string - version string + image string wantErr bool }{ { - name: "valid version", - version: "7.3.3", + name: "Official: valid version", + image: "confluentinc/cp-kafka:7.3.3", wantErr: false, }, { - name: "valid version: limit", - version: "7.0.0", + name: "Official: valid, limit version", + image: "confluentinc/cp-kafka:7.0.0", wantErr: false, }, { - name: "invalid version: low version", - version: "6.99.99", + name: "Official: invalid, low version", + image: "confluentinc/cp-kafka:6.99.99", wantErr: true, }, { - name: "invalid version: too low", - version: "5.0.0", + name: "Official: invalid, too low version", + image: "confluentinc/cp-kafka:5.0.0", wantErr: true, }, + { + name: "Unofficial does not validate KRaft version", + image: "my-kafka:1.0.0", + wantErr: false, + }, } for _, test := range tests { t.Run(test.name, func(t *testing.T) { - err := validateKRaftVersion("test-kafka:" + test.version) + err := validateKRaftVersion(test.image) if test.wantErr && err == nil { t.Fatalf("expected error, got nil") From 6df58190a465b608ea2bb49717bb2d2e4f91d153 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Manuel=20de=20la=20Pe=C3=B1a?= Date: Tue, 12 Sep 2023 12:38:24 +0200 Subject: [PATCH 13/16] docs: improve message for image validation --- docs/modules/kafka.md | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/docs/modules/kafka.md b/docs/modules/kafka.md index 147e9af6d6..efebbfa018 100644 --- a/docs/modules/kafka.md +++ b/docs/modules/kafka.md @@ -40,6 +40,11 @@ When starting the Kafka container, you can pass options in a variadic way to con If you need to set a different Kafka Docker image, you can use `testcontainers.WithImage` with a valid Docker image for Kafka. E.g. `testcontainers.WithImage("confluentinc/cp-kafka:7.3.3")`. +!!! warning + The minimal required version of Kafka for KRaft mode is `confluentinc/cp-kafka:7.0.0`. If you are using an image that + is different from the official one, please make sure that it's compatible with KRaft mode, as the module won't check + the version for you. + #### Init script The Kafka container will be started using a custom shell script: From ee2fb4828ddf587befca238cc76e24fc8ce4519f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Manuel=20de=20la=20Pe=C3=B1a?= Date: Fri, 15 Sep 2023 12:55:24 +0200 Subject: [PATCH 14/16] chore: run go mod tidy --- modules/kafka/go.mod | 3 ++- modules/kafka/go.sum | 10 +++++----- 2 files changed, 7 insertions(+), 6 deletions(-) diff --git a/modules/kafka/go.mod b/modules/kafka/go.mod index cf6ae0296d..019a247a53 100644 --- a/modules/kafka/go.mod +++ b/modules/kafka/go.mod @@ -13,8 +13,9 @@ require ( dario.cat/mergo v1.0.0 // indirect github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1 // indirect github.com/Microsoft/go-winio v0.6.1 // indirect + github.com/Microsoft/hcsshim v0.11.0 // indirect github.com/cenkalti/backoff/v4 v4.2.1 // indirect - github.com/containerd/containerd v1.7.3 // indirect + github.com/containerd/containerd v1.7.6 // indirect github.com/cpuguy83/dockercfg v0.3.1 // indirect github.com/davecgh/go-spew v1.1.1 // indirect github.com/docker/distribution v2.8.2+incompatible // indirect diff --git a/modules/kafka/go.sum b/modules/kafka/go.sum index 6fadfef92b..29a61b9abd 100644 --- a/modules/kafka/go.sum +++ b/modules/kafka/go.sum @@ -1,6 +1,6 @@ dario.cat/mergo v1.0.0 h1:AGCNq9Evsj31mOgNPcLyXc+4PNABt905YmuqPYYpBWk= dario.cat/mergo v1.0.0/go.mod h1:uNxQE+84aUszobStD9th8a29P2fMDhsBdgRYvZOxGmk= -github.com/AdaLogics/go-fuzz-headers v0.0.0-20230106234847-43070de90fa1 h1:EKPd1INOIyr5hWOWhvpmQpY6tKjeG0hT1s3AMC/9fic= +github.com/AdaLogics/go-fuzz-headers v0.0.0-20230811130428-ced1acdcaa24 h1:bvDV9vkmnHYOMsOr4WLk+Vo07yKIzd94sVoIqshQ4bU= github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1 h1:UQHMgLO+TxOElx5B5HZ4hJQsoJ/PvUvKRhJHDQXO8P8= github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1/go.mod h1:xomTg63KZ2rFqZQzSB4Vz2SUXa1BpHTVz9L5PTmPC4E= github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU= @@ -8,20 +8,20 @@ github.com/IBM/sarama v1.41.1 h1:B4/TdHce/8Ipza+qrLIeNJ9D1AOxZVp/3uDv6H/dp2M= github.com/IBM/sarama v1.41.1/go.mod h1:JFCPURVskaipJdKRFkiE/OZqQHw7jqliaJmRwXCmSSw= github.com/Microsoft/go-winio v0.6.1 h1:9/kr64B9VUZrLm5YYwbGtUJnMgqWVOdUAXu6Migciow= github.com/Microsoft/go-winio v0.6.1/go.mod h1:LRdKpFKfdobln8UmuiYcKPot9D2v6svN5+sAH+4kjUM= -github.com/Microsoft/hcsshim v0.10.0-rc.8 h1:YSZVvlIIDD1UxQpJp0h+dnpLUw+TrY0cx8obKsp3bek= +github.com/Microsoft/hcsshim v0.11.0 h1:7EFNIY4igHEXUdj1zXgAyU3fLc7QfOKHbkldRVTBdiM= +github.com/Microsoft/hcsshim v0.11.0/go.mod h1:OEthFdQv/AD2RAdzR6Mm1N1KPCztGKDurW1Z8b8VGMM= github.com/cenkalti/backoff/v4 v4.2.1 h1:y4OZtCnogmCPw98Zjyt5a6+QwPLGkiQsYW5oUqylYbM= github.com/cenkalti/backoff/v4 v4.2.1/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE= github.com/checkpoint-restore/go-criu/v5 v5.3.0/go.mod h1:E/eQpaFtUKGOOSEBZgmKAcn+zUUwWxqcaKZlF54wK8E= github.com/cilium/ebpf v0.7.0/go.mod h1:/oI2+1shJiTGAMgl6/RgJr36Eo1jzrRcAWbcXO2usCA= github.com/containerd/console v1.0.3/go.mod h1:7LqA/THxQ86k76b8c/EMSiaJ3h1eZkMkXar0TQ1gf3U= -github.com/containerd/containerd v1.7.3 h1:cKwYKkP1eTj54bP3wCdXXBymmKRQMrWjkLSWZZJDa8o= -github.com/containerd/containerd v1.7.3/go.mod h1:32FOM4/O0RkNg7AjQj3hDzN9cUGtu+HMvaKUNiqCZB8= +github.com/containerd/containerd v1.7.6 h1:oNAVsnhPoy4BTPQivLgTzI9Oleml9l/+eYIDYXRCYo8= +github.com/containerd/containerd v1.7.6/go.mod h1:SY6lrkkuJT40BVNO37tlYTSnKJnP5AXBc0fhx0q+TJ4= github.com/coreos/go-systemd/v22 v22.3.2/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc= github.com/cpuguy83/dockercfg v0.3.1 h1:/FpZ+JaygUR/lZP2NlFI2DVfrOEMAIKP5wWEJdoYe9E= github.com/cpuguy83/dockercfg v0.3.1/go.mod h1:sugsbF4//dDlL/i+S+rtpIWp+5h0BHJHfjj5/jFyUJc= github.com/cpuguy83/go-md2man/v2 v2.0.0-20190314233015-f79a8a8ca69d/go.mod h1:maD7wRr/U5Z6m/iR4s+kqSMx2CaBsrgA7czyZG/E6dU= github.com/creack/pty v1.1.18 h1:n56/Zwd5o6whRC5PMGretI4IdRLlmBXYNjScPaBgsbY= -github.com/cyphar/filepath-securejoin v0.2.3 h1:YX6ebbZCZP7VkM3scTTokDgBL2TY741X51MTk3ycuNI= github.com/cyphar/filepath-securejoin v0.2.3/go.mod h1:aPGpWjXOXUn2NCNjFvBE6aRxGGx79pTxQpKOJNYHHl4= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= From 74f10dce116e1b562d8a3ccf1d7820b0059714b0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Manuel=20de=20la=20Pe=C3=B1a?= Date: Fri, 15 Sep 2023 13:01:04 +0200 Subject: [PATCH 15/16] chore: rename file to only exists for testing --- modules/kafka/{testing.go => consumer_test.go} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename modules/kafka/{testing.go => consumer_test.go} (100%) diff --git a/modules/kafka/testing.go b/modules/kafka/consumer_test.go similarity index 100% rename from modules/kafka/testing.go rename to modules/kafka/consumer_test.go From d10d28ec06bd4fef686bd333c3e9b638dc04d700 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Manuel=20de=20la=20Pe=C3=B1a?= Date: Fri, 15 Sep 2023 13:09:09 +0200 Subject: [PATCH 16/16] chore: use concluent-local Docker image --- docs/modules/kafka.md | 4 ++-- modules/kafka/examples_test.go | 2 +- modules/kafka/kafka.go | 9 +++++---- modules/kafka/kafka_test.go | 14 +++++++------- 4 files changed, 15 insertions(+), 14 deletions(-) diff --git a/docs/modules/kafka.md b/docs/modules/kafka.md index efebbfa018..8971a2725f 100644 --- a/docs/modules/kafka.md +++ b/docs/modules/kafka.md @@ -38,10 +38,10 @@ When starting the Kafka container, you can pass options in a variadic way to con #### Image If you need to set a different Kafka Docker image, you can use `testcontainers.WithImage` with a valid Docker image -for Kafka. E.g. `testcontainers.WithImage("confluentinc/cp-kafka:7.3.3")`. +for Kafka. E.g. `testcontainers.WithImage("confluentinc/confluent-local:7.5.0")`. !!! warning - The minimal required version of Kafka for KRaft mode is `confluentinc/cp-kafka:7.0.0`. If you are using an image that + The minimal required version of Kafka for KRaft mode is `confluentinc/confluent-local:7.4.0`. If you are using an image that is different from the official one, please make sure that it's compatible with KRaft mode, as the module won't check the version for you. diff --git a/modules/kafka/examples_test.go b/modules/kafka/examples_test.go index c5a23b289a..61fe81f504 100644 --- a/modules/kafka/examples_test.go +++ b/modules/kafka/examples_test.go @@ -14,7 +14,7 @@ func ExampleRunContainer() { kafkaContainer, err := kafka.RunContainer(ctx, kafka.WithClusterID("test-cluster"), - testcontainers.WithImage("confluentinc/cp-kafka:7.3.3"), + testcontainers.WithImage("confluentinc/confluent-local:7.5.0"), ) if err != nil { panic(err) diff --git a/modules/kafka/kafka.go b/modules/kafka/kafka.go index 45fd2a5fe5..016509229b 100644 --- a/modules/kafka/kafka.go +++ b/modules/kafka/kafka.go @@ -39,11 +39,12 @@ type KafkaContainer struct { // RunContainer creates an instance of the Kafka container type func RunContainer(ctx context.Context, opts ...testcontainers.ContainerCustomizer) (*KafkaContainer, error) { req := testcontainers.ContainerRequest{ - Image: "confluentinc/cp-kafka:7.3.3", + Image: "confluentinc/confluent-local:7.5.0", ExposedPorts: []string{string(publicPort)}, Env: map[string]string{ // envVars { "KAFKA_LISTENERS": "PLAINTEXT://0.0.0.0:9093,BROKER://0.0.0.0:9092,CONTROLLER://0.0.0.0:9094", + "KAFKA_REST_BOOTSTRAP_SERVERS": "PLAINTEXT://0.0.0.0:9093,BROKER://0.0.0.0:9092,CONTROLLER://0.0.0.0:9094", "KAFKA_LISTENER_SECURITY_PROTOCOL_MAP": "BROKER:PLAINTEXT,PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT", "KAFKA_INTER_BROKER_LISTENER_NAME": "BROKER", "KAFKA_BROKER_ID": "1", @@ -169,7 +170,7 @@ func validateKRaftVersion(fqName string) error { image := fqName[:strings.LastIndex(fqName, ":")] version := fqName[strings.LastIndex(fqName, ":")+1:] - if !strings.EqualFold(image, "confluentinc/cp-kafka") { + if !strings.EqualFold(image, "confluentinc/confluent-local") { // do not validate if the image is not the official one. // not raising an error here, letting the image to start and // eventually evaluate an error if it exists. @@ -181,8 +182,8 @@ func validateKRaftVersion(fqName string) error { version = fmt.Sprintf("v%s", version) } - if semver.Compare(version, "v7.0.0") < 0 { // version < v7.0.0 - return fmt.Errorf("version=%s. KRaft mode is only available since version 7.0.0", version) + if semver.Compare(version, "v7.4.0") < 0 { // version < v7.4.0 + return fmt.Errorf("version=%s. KRaft mode is only available since version 7.4.0", version) } return nil diff --git a/modules/kafka/kafka_test.go b/modules/kafka/kafka_test.go index 4edc9095e7..821ca66fef 100644 --- a/modules/kafka/kafka_test.go +++ b/modules/kafka/kafka_test.go @@ -15,7 +15,7 @@ func TestKafka(t *testing.T) { ctx := context.Background() - kafkaContainer, err := RunContainer(ctx, WithClusterID("kraftCluster"), testcontainers.WithImage("confluentinc/cp-kafka:7.3.3")) + kafkaContainer, err := RunContainer(ctx, WithClusterID("kraftCluster"), testcontainers.WithImage("confluentinc/confluent-local:7.5.0")) if err != nil { t.Fatal(err) } @@ -87,7 +87,7 @@ func TestKafka(t *testing.T) { func TestKafka_invalidVersion(t *testing.T) { ctx := context.Background() - _, err := RunContainer(ctx, WithClusterID("kraftCluster"), testcontainers.WithImage("confluentinc/cp-kafka:6.3.3")) + _, err := RunContainer(ctx, WithClusterID("kraftCluster"), testcontainers.WithImage("confluentinc/confluent-local:6.3.3")) if err == nil { t.Fatal(err) } @@ -149,7 +149,7 @@ func TestConfigureQuorumVoters(t *testing.T) { } } -func TestVAlidateKRaftVersion(t *testing.T) { +func TestValidateKRaftVersion(t *testing.T) { tests := []struct { name string image string @@ -157,22 +157,22 @@ func TestVAlidateKRaftVersion(t *testing.T) { }{ { name: "Official: valid version", - image: "confluentinc/cp-kafka:7.3.3", + image: "confluentinc/confluent-local:7.5.0", wantErr: false, }, { name: "Official: valid, limit version", - image: "confluentinc/cp-kafka:7.0.0", + image: "confluentinc/confluent-local:7.4.0", wantErr: false, }, { name: "Official: invalid, low version", - image: "confluentinc/cp-kafka:6.99.99", + image: "confluentinc/confluent-local:7.3.99", wantErr: true, }, { name: "Official: invalid, too low version", - image: "confluentinc/cp-kafka:5.0.0", + image: "confluentinc/confluent-local:5.0.0", wantErr: true, }, {