Skip to content

Commit 1ff166c

Browse files
ghthorghthor
committed
feat: additional advertised listners/broker addresses for kafka
1 parent 35bf0cd commit 1ff166c

File tree

3 files changed

+511
-21
lines changed

3 files changed

+511
-21
lines changed

‎modules/kafka/examples_test.go‎

Lines changed: 297 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,12 +1,19 @@
11
package kafka_test
22

33
import (
4+
"bytes"
45
"context"
56
"fmt"
7+
"io"
68
"log"
9+
"strings"
710

11+
"github.com/IBM/sarama"
12+
"github.com/docker/docker/api/types/container"
813
"github.com/testcontainers/testcontainers-go"
914
"github.com/testcontainers/testcontainers-go/modules/kafka"
15+
"github.com/testcontainers/testcontainers-go/network"
16+
"github.com/testcontainers/testcontainers-go/wait"
1017
)
1118

1219
func ExampleRun() {
@@ -41,3 +48,293 @@ func ExampleRun() {
4148
// test-cluster
4249
// true
4350
}
51+
52+
func ExampleKafkaContainer_BrokersByHostDockerInternal() {
53+
ctx := context.Background()
54+
55+
kafkaContainer, err := kafka.Run(ctx,
56+
"confluentinc/confluent-local:7.5.0",
57+
kafka.WithClusterID("test-cluster"),
58+
)
59+
if err != nil {
60+
log.Fatalf("failed to start container: %s", err)
61+
}
62+
63+
// Clean up the container after
64+
defer func() {
65+
if err := kafkaContainer.Terminate(ctx); err != nil {
66+
log.Fatalf("failed to terminate container: %s", err)
67+
}
68+
}()
69+
70+
{
71+
state, err := kafkaContainer.State(ctx)
72+
if err != nil {
73+
log.Fatalf("failed to get container state: %s", err) // nolint:gocritic
74+
}
75+
76+
fmt.Println(kafkaContainer.ClusterID)
77+
fmt.Println(state.Running)
78+
}
79+
80+
const topic = "example-topic"
81+
82+
// Produce a message from the host that will be read by a consumer in another docker container
83+
{
84+
brokers, err := kafkaContainer.Brokers(ctx)
85+
86+
config := sarama.NewConfig()
87+
config.Producer.Return.Successes = true
88+
producer, err := sarama.NewSyncProducer(brokers, config)
89+
if err != nil {
90+
log.Fatal(err)
91+
}
92+
93+
if _, _, err := producer.SendMessage(&sarama.ProducerMessage{
94+
Topic: topic,
95+
Key: sarama.StringEncoder("key"),
96+
Value: sarama.StringEncoder("example_message_value"),
97+
}); err != nil {
98+
log.Fatal(err)
99+
}
100+
101+
}
102+
103+
brokers, err := kafkaContainer.BrokersByHostDockerInternal(ctx)
104+
if err != nil {
105+
log.Fatal(err)
106+
}
107+
108+
// Run another container that can connect to the kafka container via hostname "host.docker.internal"
109+
kcat, err := testcontainers.GenericContainer(
110+
ctx,
111+
testcontainers.GenericContainerRequest{
112+
ContainerRequest: testcontainers.ContainerRequest{
113+
Image: "confluentinc/cp-kafkacat",
114+
Entrypoint: []string{"kafkacat"},
115+
Cmd: []string{"-b", strings.Join(brokers, ","), "-C", "-t", topic, "-c", "1"},
116+
WaitingFor: wait.ForExit(),
117+
118+
// Add host.docker.internal to the consumer container so it can contact the kafka borkers
119+
HostConfigModifier: func(hc *container.HostConfig) {
120+
hc.ExtraHosts = append(hc.ExtraHosts, "host.docker.internal:host-gateway")
121+
},
122+
},
123+
Started: true,
124+
},
125+
)
126+
if err != nil {
127+
log.Fatalf("kafkacat error: %v", err)
128+
}
129+
130+
lr, err := kcat.Logs(ctx)
131+
if err != nil {
132+
log.Fatalf("kafkacat logs error: %v", err)
133+
}
134+
135+
logs, err := io.ReadAll(lr)
136+
if err != nil {
137+
log.Fatalf("kafkacat logs read error: %v", err)
138+
}
139+
140+
fmt.Println("read message:", string(bytes.TrimSpace(logs)))
141+
142+
// Output:
143+
// test-cluster
144+
// true
145+
// read message: example_message_value
146+
}
147+
148+
func ExampleKafkaContainer_BrokersByContainerName() {
149+
ctx := context.Background()
150+
151+
net, err := network.New(ctx)
152+
if err != nil {
153+
log.Fatalf("failed to create network: %s", err)
154+
}
155+
156+
kafkaContainer, err := kafka.Run(ctx,
157+
"confluentinc/confluent-local:7.5.0",
158+
kafka.WithClusterID("test-cluster"),
159+
network.WithNetwork(nil, net), // Run kafka test container in a new docker network
160+
)
161+
if err != nil {
162+
log.Fatalf("failed to start container: %s", err)
163+
}
164+
165+
// Clean up the container after
166+
defer func() {
167+
if err := kafkaContainer.Terminate(ctx); err != nil {
168+
log.Fatalf("failed to terminate container: %s", err)
169+
}
170+
}()
171+
172+
{
173+
state, err := kafkaContainer.State(ctx)
174+
if err != nil {
175+
log.Fatalf("failed to get container state: %s", err) // nolint:gocritic
176+
}
177+
178+
fmt.Println(kafkaContainer.ClusterID)
179+
fmt.Println(state.Running)
180+
}
181+
182+
const topic = "example-topic"
183+
184+
// Produce a message from the host that will be read by a consumer in another docker container
185+
{
186+
brokers, err := kafkaContainer.Brokers(ctx)
187+
188+
config := sarama.NewConfig()
189+
config.Producer.Return.Successes = true
190+
producer, err := sarama.NewSyncProducer(brokers, config)
191+
if err != nil {
192+
log.Fatal(err)
193+
}
194+
195+
if _, _, err := producer.SendMessage(&sarama.ProducerMessage{
196+
Topic: topic,
197+
Key: sarama.StringEncoder("key"),
198+
Value: sarama.StringEncoder("example_message_value"),
199+
}); err != nil {
200+
log.Fatal(err)
201+
}
202+
}
203+
204+
brokers, err := kafkaContainer.BrokersByContainerName(ctx)
205+
if err != nil {
206+
log.Fatal(err)
207+
}
208+
209+
// Run another container that can connect to the kafka container via the kafka containers name
210+
kcat, err := testcontainers.GenericContainer(
211+
ctx,
212+
testcontainers.GenericContainerRequest{
213+
ContainerRequest: testcontainers.ContainerRequest{
214+
Image: "confluentinc/cp-kafkacat",
215+
Entrypoint: []string{"kafkacat"},
216+
Cmd: []string{"-b", strings.Join(brokers, ","), "-C", "-t", topic, "-c", "1"},
217+
WaitingFor: wait.ForExit(),
218+
Networks: []string{net.Name}, // Run kafkacat in the same docker network as the testcontainer
219+
},
220+
Started: true,
221+
},
222+
)
223+
if err != nil {
224+
log.Fatalf("kafkacat error: %v", err)
225+
}
226+
227+
lr, err := kcat.Logs(ctx)
228+
if err != nil {
229+
log.Fatalf("kafkacat logs error: %v", err)
230+
}
231+
232+
logs, err := io.ReadAll(lr)
233+
if err != nil {
234+
log.Fatalf("kafkacat logs read error: %v", err)
235+
}
236+
237+
fmt.Println("read message:", string(bytes.TrimSpace(logs)))
238+
239+
// Output:
240+
// test-cluster
241+
// true
242+
// read message: example_message_value
243+
}
244+
245+
func ExampleKafkaContainer_BrokersByContainerId() {
246+
ctx := context.Background()
247+
248+
net, err := network.New(ctx)
249+
if err != nil {
250+
log.Fatalf("failed to create network: %s", err)
251+
}
252+
253+
kafkaContainer, err := kafka.Run(ctx,
254+
"confluentinc/confluent-local:7.5.0",
255+
kafka.WithClusterID("test-cluster"),
256+
network.WithNetwork(nil, net), // Run kafka test container in a new docker network
257+
)
258+
if err != nil {
259+
log.Fatalf("failed to start container: %s", err)
260+
}
261+
262+
// Clean up the container after
263+
defer func() {
264+
if err := kafkaContainer.Terminate(ctx); err != nil {
265+
log.Fatalf("failed to terminate container: %s", err)
266+
}
267+
}()
268+
269+
{
270+
state, err := kafkaContainer.State(ctx)
271+
if err != nil {
272+
log.Fatalf("failed to get container state: %s", err) // nolint:gocritic
273+
}
274+
275+
fmt.Println(kafkaContainer.ClusterID)
276+
fmt.Println(state.Running)
277+
}
278+
279+
const topic = "example-topic"
280+
281+
// Produce a message from the host that will be read by a consumer in another docker container
282+
{
283+
brokers, err := kafkaContainer.Brokers(ctx)
284+
285+
config := sarama.NewConfig()
286+
config.Producer.Return.Successes = true
287+
producer, err := sarama.NewSyncProducer(brokers, config)
288+
if err != nil {
289+
log.Fatal(err)
290+
}
291+
292+
if _, _, err := producer.SendMessage(&sarama.ProducerMessage{
293+
Topic: topic,
294+
Key: sarama.StringEncoder("key"),
295+
Value: sarama.StringEncoder("example_message_value"),
296+
}); err != nil {
297+
log.Fatal(err)
298+
}
299+
}
300+
301+
brokers, err := kafkaContainer.BrokersByContainerId(ctx)
302+
if err != nil {
303+
log.Fatal(err)
304+
}
305+
306+
// Run another container that can connect to the kafka container via the kafka containers ContainerID
307+
kcat, err := testcontainers.GenericContainer(
308+
ctx,
309+
testcontainers.GenericContainerRequest{
310+
ContainerRequest: testcontainers.ContainerRequest{
311+
Image: "confluentinc/cp-kafkacat",
312+
Entrypoint: []string{"kafkacat"},
313+
Cmd: []string{"-b", strings.Join(brokers, ","), "-C", "-t", topic, "-c", "1"},
314+
WaitingFor: wait.ForExit(),
315+
Networks: []string{net.Name}, // Run kafkacat in the same docker network as the testcontainer
316+
},
317+
Started: true,
318+
},
319+
)
320+
if err != nil {
321+
log.Fatalf("kafkacat error: %v", err)
322+
}
323+
324+
lr, err := kcat.Logs(ctx)
325+
if err != nil {
326+
log.Fatalf("kafkacat logs error: %v", err)
327+
}
328+
329+
logs, err := io.ReadAll(lr)
330+
if err != nil {
331+
log.Fatalf("kafkacat logs read error: %v", err)
332+
}
333+
334+
fmt.Println("read message:", string(bytes.TrimSpace(logs)))
335+
336+
// Output:
337+
// test-cluster
338+
// true
339+
// read message: example_message_value
340+
}

0 commit comments

Comments
 (0)