Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Use new apache/kafka-native docker image for kafka module #2683

Draft
wants to merge 3 commits into
base: main
Choose a base branch
from
+47 −191
Draft

Use new apache/kafka-native docker image for kafka module #2683

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
6822e77
getting some tests passing
CTrando Aug 5, 2024
cc64b73
something working for now
CTrando Aug 5, 2024
b55139d
no need to include that
CTrando Aug 5, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion modules/kafka/examples_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@ func ExampleRun() {
ctx := context.Background()

kafkaContainer, err := kafka.Run(ctx,
"confluentinc/confluent-local:7.5.0",
"apache/kafka-native:3.8.0",
kafka.WithClusterID("test-cluster"),
)
if err != nil {
Expand Down
142 changes: 45 additions & 97 deletions modules/kafka/kafka.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,31 +3,17 @@ package kafka
import (
"context"
"fmt"
"math"
"strings"
"github.com/docker/docker/api/types/container"
"github.com/testcontainers/testcontainers-go/wait"
"strconv"
"time"

"github.com/docker/go-connections/nat"
"golang.org/x/mod/semver"

"github.com/testcontainers/testcontainers-go"
"github.com/testcontainers/testcontainers-go/wait"
)

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
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`
// }
publicPort = nat.Port("9093/tcp")
)

// KafkaContainer represents the Kafka container type used in the module
Expand All @@ -39,67 +25,49 @@ type KafkaContainer struct {
// Deprecated: use Run instead
// RunContainer creates an instance of the Kafka container type
func RunContainer(ctx context.Context, opts ...testcontainers.ContainerCustomizer) (*KafkaContainer, error) {
return Run(ctx, "confluentinc/confluent-local:7.5.0", opts...)
return Run(ctx, "apache/kafka-native:3.8.0", opts...)
}

// Run creates an instance of the Kafka container type
func Run(ctx context.Context, img string, opts ...testcontainers.ContainerCustomizer) (*KafkaContainer, error) {
req := testcontainers.ContainerRequest{
Image: img,
ExposedPorts: []string{string(publicPort)},
WaitingFor: wait.ForLog("Transition from STARTING to STARTED (kafka." +
"server.BrokerServer)").WithStartupTimeout(2 * time.Minute),
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_LISTENERS": "CONTROLLER://0.0.0.0:9094,PLAINTEXT://0.0.0.0:9093,BROKER://0.0.0.0:9092",
"KAFKA_ADVERTISED_LISTENERS": "PLAINTEXT://localhost:9093,BROKER://localhost:9092",
"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": "CONTROLLER" +
":PLAINTEXT,SASL_PLAINTEXT:SASL_PLAINTEXT," +
"PLAINTEXT:PLAINTEXT," +
"BROKER:PLAINTEXT",
"KAFKA_CONTROLLER_LISTENER_NAMES": "CONTROLLER",
"KAFKA_BROKER_ID": "1",
"KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR": "1",
"KAFKA_OFFSETS_TOPIC_NUM_PARTITIONS": "1",
"KAFKA_INTER_BROKER_LISTENER_NAME": "BROKER",
"KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL": "PLAIN",
"KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR": "1",
"KAFKA_TRANSACTION_STATE_LOG_MIN_ISR": "1",
"KAFKA_LOG_FLUSH_INTERVAL_MESSAGES": fmt.Sprintf("%d", math.MaxInt),
"KAFKA_LOG_FLUSH_INTERVAL_MESSAGES": "9223372036854775807", // math.MaxInt value
"KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS": "0",
"KAFKA_NODE_ID": "1",
"KAFKA_PROCESS_ROLES": "broker,controller",
"KAFKA_CONTROLLER_LISTENER_NAMES": "CONTROLLER",
// }
"KAFKA_CONTROLLER_QUORUM_VOTERS": "1@localhost:9094",
"KAFKA_SASL_ENABLED_MECHANISMS": "PLAIN",
"KAFKA_OPTS": " ",
},
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
}

inspect, err := c.Inspect(ctx)
if err != nil {
return err
}

hostname := inspect.Config.Hostname

port, err := c.MappedPort(ctx, publicPort)
if err != nil {
return err
}

scriptContent := fmt.Sprintf(starterScriptContent, host, port.Int(), hostname)

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(".*Transitioning from RECOVERY to RUNNING.*").AsRegexp().WaitUntilReady(ctx, c)
HostConfigModifier: func(config *container.HostConfig) {
config.PortBindings = map[nat.Port][]nat.PortBinding{
"9093/tcp": {
{
HostIP: "0.0.0.0",
HostPort: "9093",
},
},
},
}
},
}

Expand All @@ -114,13 +82,7 @@ func Run(ctx context.Context, img string, opts ...testcontainers.ContainerCustom
}
}

err := validateKRaftVersion(genericContainerReq.Image)
if err != nil {
return nil, err
}

clusterID := genericContainerReq.Env["CLUSTER_ID"]

configureControllerQuorumVoters(&genericContainerReq)

container, err := testcontainers.GenericContainer(ctx, genericContainerReq)
Expand All @@ -139,6 +101,22 @@ func WithClusterID(clusterID string) testcontainers.CustomizeRequestOption {
}
}

func WithHostPort(port int) testcontainers.CustomizeRequestOption {
return func(req *testcontainers.GenericContainerRequest) error {
req.HostConfigModifier = func(config *container.HostConfig) {
config.PortBindings = map[nat.Port][]nat.PortBinding{
"9093/tcp": {
{
HostIP: "0.0.0.0",
HostPort: strconv.Itoa(port),
},
},
}
}
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) {
Expand Down Expand Up @@ -174,34 +152,4 @@ func configureControllerQuorumVoters(req *testcontainers.GenericContainerRequest

req.Env["KAFKA_CONTROLLER_QUORUM_VOTERS"] = fmt.Sprintf("1@%s:9094", host)
}
// }
}

// validateKRaftVersion validates if the image version is compatible with KRaft mode,
// which is available since version 7.0.0.
func validateKRaftVersion(fqName string) error {
Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

every apache/kafka-native image supports kraft

if fqName == "" {
return fmt.Errorf("image cannot be empty")
}

image := fqName[:strings.LastIndex(fqName, ":")]
version := fqName[strings.LastIndex(fqName, ":")+1:]

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.
return nil
}

// 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.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
}
48 changes: 0 additions & 48 deletions modules/kafka/kafka_helpers_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,51 +61,3 @@ func TestConfigureQuorumVoters(t *testing.T) {
})
}
}

func TestValidateKRaftVersion(t *testing.T) {
tests := []struct {
name string
image string
wantErr bool
}{
{
name: "Official: valid version",
image: "confluentinc/confluent-local:7.5.0",
wantErr: false,
},
{
name: "Official: valid, limit version",
image: "confluentinc/confluent-local:7.4.0",
wantErr: false,
},
{
name: "Official: invalid, low version",
image: "confluentinc/confluent-local:7.3.99",
wantErr: true,
},
{
name: "Official: invalid, too low version",
image: "confluentinc/confluent-local: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.image)

if test.wantErr && err == nil {
t.Fatalf("expected error, got nil")
}

if !test.wantErr && err != nil {
t.Fatalf("expected no error, got %s", err)
}
})
}
}
46 changes: 1 addition & 45 deletions modules/kafka/kafka_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,13 +2,11 @@ package kafka_test

import (
"context"
"io"
"strings"
"testing"

"github.com/IBM/sarama"

"github.com/testcontainers/testcontainers-go"
"github.com/testcontainers/testcontainers-go/modules/kafka"
)

Expand All @@ -17,7 +15,7 @@ func TestKafka(t *testing.T) {

ctx := context.Background()

kafkaContainer, err := kafka.Run(ctx, "confluentinc/confluent-local:7.5.0", kafka.WithClusterID("kraftCluster"))
kafkaContainer, err := kafka.Run(ctx, "apache/kafka-native:3.8.0", kafka.WithClusterID("kraftCluster"))
Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

TODO: need some more tests on the client side to demonstrate this works + add tests for setting host port

if err != nil {
t.Fatal(err)
}
Expand All @@ -29,8 +27,6 @@ func TestKafka(t *testing.T) {
}
})

assertAdvertisedListeners(t, kafkaContainer)

if !strings.EqualFold(kafkaContainer.ClusterID, "kraftCluster") {
t.Fatalf("expected clusterID to be %s, got %s", "kraftCluster", kafkaContainer.ClusterID)
}
Expand Down Expand Up @@ -58,8 +54,6 @@ func TestKafka(t *testing.T) {
// 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

Expand Down Expand Up @@ -87,41 +81,3 @@ func TestKafka(t *testing.T) {
t.Fatalf("expected value to be %s, got %s", "value", string(consumer.message.Value))
}
}

func TestKafka_invalidVersion(t *testing.T) {
ctx := context.Background()

_, err := kafka.Run(ctx, "confluentinc/confluent-local:6.3.3", kafka.WithClusterID("kraftCluster"))
if err == nil {
t.Fatal(err)
}
}

// assertAdvertisedListeners checks that the advertised listeners are set correctly:
// - The BROKER:// protocol is using the hostname of the Kafka container
func assertAdvertisedListeners(t *testing.T, container testcontainers.Container) {
inspect, err := container.Inspect(context.Background())
if err != nil {
t.Fatal(err)
}

hostname := inspect.Config.Hostname

code, r, err := container.Exec(context.Background(), []string{"cat", "/usr/sbin/testcontainers_start.sh"})
Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this script doesn't exist anymore in the kafka image - also I didn't think it was necessary to check advertisted listeners, considering if it's set incorrectly, the other tests would fail because kafka wouldn't work in the first place

if err != nil {
t.Fatal(err)
}

if code != 0 {
t.Fatalf("expected exit code to be 0, got %d", code)
}

bs, err := io.ReadAll(r)
if err != nil {
t.Fatal(err)
}

if !strings.Contains(string(bs), "BROKER://"+hostname+":9092") {
t.Fatalf("expected advertised listeners to contain %s, got %s", "BROKER://"+hostname+":9092", string(bs))
}
}