fix(kafka): Fix internal docker connection #2894
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -3,7 +3,7 @@ module github.com/testcontainers/testcontainers-go/modules/kafka | |
| go 1.22 | ||
|
|
||
| require ( | ||
| github.com/IBM/sarama v1.43.3 | ||
| github.com/docker/go-connections v0.5.0 | ||
| github.com/stretchr/testify v1.9.0 | ||
| github.com/testcontainers/testcontainers-go v0.34.0 | ||
|
|
@@ -23,7 +23,7 @@ require ( | |
| github.com/distribution/reference v0.6.0 // indirect | ||
| github.com/docker/docker v27.1.1+incompatible // indirect | ||
| github.com/docker/go-units v0.5.0 // indirect | ||
| github.com/eapache/go-resiliency v1.7.0 // indirect | ||
| github.com/eapache/go-xerial-snappy v0.0.0-20230731223053-c322873962e3 // indirect | ||
| github.com/eapache/queue v1.1.0 // indirect | ||
| github.com/felixge/httpsnoop v1.0.4 // indirect | ||
|
|
@@ -41,7 +41,7 @@ require ( | |
| 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.17.9 // indirect | ||
| github.com/kr/text v0.2.0 // indirect | ||
| github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect | ||
| github.com/magiconair/properties v1.8.7 // indirect | ||
|
|
@@ -53,7 +53,7 @@ require ( | |
| 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 // indirect | ||
| github.com/pierrec/lz4/v4 v4.1.21 // indirect | ||
| github.com/pkg/errors v0.9.1 // indirect | ||
| github.com/pmezard/go-difflib v1.0.0 // indirect | ||
| github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect | ||
|
|
@@ -69,8 +69,7 @@ require ( | |
| go.opentelemetry.io/otel/metric v1.24.0 // indirect | ||
| go.opentelemetry.io/otel/trace v1.24.0 // indirect | ||
| golang.org/x/crypto v0.28.0 // indirect | ||
| golang.org/x/net v0.28.0 // indirect | ||
| golang.org/x/sys v0.26.0 // indirect | ||
| google.golang.org/genproto/googleapis/api v0.0.0-20240318140521-94a12d6c2237 // indirect | ||
| google.golang.org/genproto/googleapis/rpc v0.0.0-20240318140521-94a12d6c2237 // indirect | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -4,8 +4,8 @@ github.com/AdaLogics/go-fuzz-headers v0.0.0-20230811130428-ced1acdcaa24 h1:bvDV9 | |
| github.com/AdaLogics/go-fuzz-headers v0.0.0-20230811130428-ced1acdcaa24/go.mod h1:8o94RPi1/7XTJvwPpRSzSUedZrtlirdB3r9Z20bi2f8= | ||
| 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/IBM/sarama v1.43.3 h1:Yj6L2IaNvb2mRBop39N7mmJAHBVY3dTPncr3qGVkxPA= | ||
| github.com/IBM/sarama v1.43.3/go.mod h1:FVIRaLrhK3Cla/9FfRF5X9Zua2KpS3SYIXxhac1H+FQ= | ||
| github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERoyfY= | ||
| github.com/Microsoft/go-winio v0.6.2/go.mod h1:yd8OoFMLzJbo9gZq8j5qaps8bJ9aShtEA8Ipt1oGCvU= | ||
| github.com/cenkalti/backoff/v4 v4.2.1 h1:y4OZtCnogmCPw98Zjyt5a6+QwPLGkiQsYW5oUqylYbM= | ||
|
|
@@ -32,8 +32,8 @@ github.com/docker/go-connections v0.5.0 h1:USnMq7hx7gwdVZq1L49hLXaFtUdTADjXGp+uj | |
| github.com/docker/go-connections v0.5.0/go.mod h1:ov60Kzw0kKElRwhNs9UlUHAE/F9Fe6GLaXnqyDdmEXc= | ||
| 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.7.0 h1:n3NRTnBn5N0Cbi/IeOHuQn9s2UwVUH7Ga0ZWcP+9JTA= | ||
| github.com/eapache/go-resiliency v1.7.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= | ||
|
|
@@ -85,8 +85,8 @@ github.com/jcmturner/rpc/v2 v2.0.3 h1:7FXXj8Ti1IaVFpSAziCZWNzbNuZmnvw/i6CqLNdWfZ | |
| 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.17.9 h1:6KIumPrER1LHsvBVuDa0r5xaG0Es51mhhB9BQB2qeMA= | ||
| github.com/klauspost/compress v1.17.9/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw= | ||
| github.com/kr/pretty v0.3.0 h1:WgNl7dwNpEZ6jJ9k1snq4pZsg7DOEN8hP9Xw0Tsjwk0= | ||
| github.com/kr/pretty v0.3.0/go.mod h1:640gp4NfQd8pI5XOwp5fnNeVWj67G7CFk/SaSQn7NBk= | ||
| github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= | ||
|
|
@@ -111,8 +111,8 @@ github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8 | |
| github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM= | ||
| github.com/opencontainers/image-spec v1.1.0 h1:8SG7/vwALn54lVB/0yZ/MMwhFrPYtpEHQb2IpWsCzug= | ||
| github.com/opencontainers/image-spec v1.1.0/go.mod h1:W4s4sFTMaBeK1BQLXbG4AdM2szdn85PY75RI83NrTrM= | ||
| github.com/pierrec/lz4/v4 v4.1.21 h1:yOVMLb6qSIDP67pl/5F7RepeKYu/VmTyEXvuMI5d9mQ= | ||
| github.com/pierrec/lz4/v4 v4.1.21/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= | ||
|
|
@@ -190,14 +190,14 @@ golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v | |
| 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.28.0 h1:a9JDOJc5GMUJ0+UDqmLT86WiEy7iWyIhz8gz8E4e5hE= | ||
| golang.org/x/net v0.28.0/go.mod h1:yqtgsTWOOnlGLG9GFRrK3++bGOUEkNBoHZc8MEDWPNg= | ||
| 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.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= | ||
| golang.org/x/sync v0.8.0 h1:3NFvSEYkUoMifnESzZl15y791HH1qU2xm6eCJU5ZPXQ= | ||
| golang.org/x/sync v0.8.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= | ||
| 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-20190916202348-b4ddaad3f8a3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -22,7 +22,7 @@ const ( | |
| // starterScript { | ||
| starterScriptContent = `#!/bin/bash | ||
| source /etc/confluent/docker/bash-config | ||
| export KAFKA_ADVERTISED_LISTENERS=%s | ||
| 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 | ||
|
|
@@ -38,6 +38,12 @@ type KafkaContainer struct { | |
| ClusterID string | ||
| } | ||
|
|
||
| type Listener struct { | ||
|
There was a problem hiding this comment. suggestion: Add doc comment. |
||
| Name string | ||
| Host string | ||
| Port string | ||
| } | ||
|
|
||
| // Deprecated: use Run instead | ||
| // RunContainer creates an instance of the Kafka container type | ||
| func RunContainer(ctx context.Context, opts ...testcontainers.ContainerCustomizer) (*KafkaContainer, error) { | ||
|
|
@@ -70,37 +76,44 @@ func Run(ctx context.Context, img string, opts ...testcontainers.ContainerCustom | |
| 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}, | ||
| } | ||
|
|
||
| genericContainerReq := testcontainers.GenericContainerRequest{ | ||
| ContainerRequest: req, | ||
| Started: true, | ||
| } | ||
|
|
||
| settings := defaultOptions(&genericContainerReq) | ||
| for _, opt := range opts { | ||
| if apply, ok := opt.(Option); ok { | ||
| if err := apply(&settings); err != nil { | ||
| return nil, err | ||
| } | ||
| } | ||
| if err := opt.Customize(&genericContainerReq); err != nil { | ||
| return nil, err | ||
| } | ||
| } | ||
|
|
||
| genericContainerReq.ContainerRequest.LifecycleHooks = []testcontainers.ContainerLifecycleHooks{ | ||
| { | ||
| PostStarts: []testcontainers.ContainerHook{ | ||
| // Use a single hook to copy the starter script and wait for | ||
| // the Kafka server to be ready. This prevents the wait running | ||
| // if the starter script fails to copy. | ||
| func(ctx context.Context, c testcontainers.Container) error { | ||
| // 1. copy the starter script into the container | ||
| if err := copyStarterScript(ctx, c, &settings); err != nil { | ||
| return fmt.Errorf("copy starter script: %w", err) | ||
| } | ||
|
|
||
| // 2. wait for the Kafka server to be ready | ||
| return wait.ForLog(".*Transitioning from RECOVERY to RUNNING.*").AsRegexp().WaitUntilReady(ctx, c) | ||
| }, | ||
| }, | ||
| }, | ||
| } | ||
|
|
||
| err := validateKRaftVersion(genericContainerReq.Image) | ||
| if err != nil { | ||
| return nil, err | ||
|
|
@@ -122,31 +135,34 @@ func Run(ctx context.Context, img string, opts ...testcontainers.ContainerCustom | |
| } | ||
|
|
||
| // copyStarterScript copies the starter script into the container. | ||
| func copyStarterScript(ctx context.Context, c testcontainers.Container, settings *options) error { | ||
| if err := wait.ForListeningPort(publicPort). | ||
| SkipInternalCheck(). | ||
| WaitUntilReady(ctx, c); err != nil { | ||
| return fmt.Errorf("wait for exposed port: %w", err) | ||
| } | ||
|
|
||
| if len(settings.Listeners) == 0 { | ||
| defaultInternal, err := brokerListener(ctx, c) | ||
| if err != nil { | ||
| return fmt.Errorf("default internal listener: %w", err) | ||
| } | ||
| settings.Listeners = append(settings.Listeners, defaultInternal) | ||
| } | ||
|
|
||
| defaultExternal, err := plainTextListener(ctx, c) | ||
| if err != nil { | ||
| return fmt.Errorf("default external listener: %w", err) | ||
| } | ||
|
|
||
| settings.Listeners = append(settings.Listeners, defaultExternal) | ||
|
|
||
| advertised := make([]string, len(settings.Listeners)) | ||
| for i, item := range settings.Listeners { | ||
| advertised[i] = item.Name + "://" + item.Host + ":" + item.Port | ||
| } | ||
|
|
||
| scriptContent := fmt.Sprintf(starterScriptContent, strings.Join(advertised, ",")) | ||
|
|
||
| if err := c.CopyToContainer(ctx, []byte(scriptContent), starterScript, 0o755); err != nil { | ||
| return fmt.Errorf("copy to container: %w", err) | ||
|
|
@@ -155,14 +171,6 @@ func copyStarterScript(ctx context.Context, c testcontainers.Container) error { | |
| return nil | ||
| } | ||
|
|
||
| // 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) { | ||
|
|
||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
suggestion: its idiomatic to use a
struct{}for this pattern