Package io.strimzi.test.container

Class StrimziKafkaContainer

java.lang.Object

org.testcontainers.containers.GenericContainer<StrimziKafkaContainer>

io.strimzi.test.container.StrimziKafkaContainer

All Implemented Interfaces:

KafkaContainer, AutoCloseable, org.testcontainers.containers.Container<StrimziKafkaContainer>, org.testcontainers.containers.ContainerState, org.testcontainers.containers.traits.LinkableContainer, org.testcontainers.containers.wait.strategy.WaitStrategyTarget, org.testcontainers.lifecycle.Startable

public class StrimziKafkaContainer
extends org.testcontainers.containers.GenericContainer<StrimziKafkaContainer>
implements KafkaContainer

StrimziKafkaContainer is a single-node instance of Kafka using the image from quay.io/strimzi/kafka with the given version.

Optionally, you can configure a proxyContainer to simulate network conditions (i.e. connection cut, latency). This class uses getBootstrapServers() to build the KAFKA_ADVERTISED_LISTENERS configuration. When proxyContainer is configured, the bootstrap URL returned by getBootstrapServers() contains the proxy host and port. For this reason, Kafka clients will always pass through the proxy, even after refreshing cluster metadata.

Nested Class Summary

Nested classes/interfaces inherited from interface org.testcontainers.containers.Container

org.testcontainers.containers.Container.ExecResult

Field Summary

Fields

Modifier and Type	Field	Description
static final int	CONTROLLER_PORT	Default Kafka controller port
static final int	KAFKA_PORT	Default Kafka port
static final String	STARTER_SCRIPT	The file containing the startup script.

Fields inherited from class org.testcontainers.containers.GenericContainer

CONTAINER_RUNNING_TIMEOUT_SEC, INTERNAL_HOST_HOSTNAME

Fields inherited from interface org.testcontainers.containers.ContainerState

STATE_HEALTHY

Method Summary

Modifier and Type	Method	Description
AuthenticationType	getAuthenticationType()	Gets the authentication type configured for this container.
String	getBootstrapControllers()	Retrieves the bootstrap controllers URL for admin clients that need to connect to controllers.
String	getBootstrapServers()	Retrieves the bootstrap servers URL for Kafka clients.
String	getClientId()	Gets the OAuth client ID.
String	getClientSecret()	Gets the OAuth client secret.
String	getClusterId()	Get the cluster id.
String	getNetworkBootstrapControllers()	Get the bootstrap controllers that containers on the same network should use to connect to controllers
String	getNetworkBootstrapServers()	Get the bootstrap servers that containers on the same network should use to connect
int	getNodeId()	Gets the node ID for this Kafka container.
KafkaNodeRole	getNodeRole()	Gets the node role for this Kafka container.
String	getOauthUri()	Gets the OAuth URI.
eu.rekawek.toxiproxy.Proxy	getProxy()	Retrieves the Proxy instance for this Kafka broker.
String	getRealm()	Gets the OAuth realm.
String	getSaslPassword()	Gets the SASL password for authentication.
String	getSaslUsername()	Gets the SASL username for authentication.
String	getUsernameClaim()	Gets the OAuth username claim.
boolean	isOAuthEnabled()	Checks if OAuth is enabled.
void	stop()
StrimziKafkaContainer	waitForRunning()	Fluent method, which sets a waiting strategy to wait until the node is ready.
StrimziKafkaContainer	withAuthenticationType(AuthenticationType authType)	Sets the authentication type for the Kafka container.
StrimziKafkaContainer	withBootstrapServers(Function<StrimziKafkaContainer,String> provider)	Fluent method, assign provider for overriding bootstrap servers string
StrimziKafkaContainer	withKafkaConfigurationMap(Map<String,String> kafkaConfigurationMap)	Fluent method, which sets kafkaConfigurationMap.
StrimziKafkaContainer	withKafkaLog(org.apache.logging.log4j.Level level)	Configures the Kafka container to use the specified logging level for Kafka logs and the io.strimzi logger.
StrimziKafkaContainer	withKafkaVersion(String kafkaVersion)	Fluent method, which sets kafkaVersion.
StrimziKafkaContainer	withLogCollection()	Fluent method to enable log collection with a default log file path.
StrimziKafkaContainer	withLogCollection(String logFilePath)	Fluent method to enable log collection for all containers in the cluster with a custom log file path.
StrimziKafkaContainer	withNodeId(int nodeId)	Fluent method that sets the node ID.
StrimziKafkaContainer	withNodeRole(KafkaNodeRole nodeRole)	Fluent method to set the Kafka node role.
StrimziKafkaContainer	withOAuthConfig(String realm, String clientId, String clientSecret, String oauthUri, String usernameClaim)	Fluent method to configure OAuth settings.
StrimziKafkaContainer	withPort(int fixedPort)	Fluent method, which sets fixed exposed port.
StrimziKafkaContainer	withProxyContainer(org.testcontainers.toxiproxy.ToxiproxyContainer proxyContainer)	Fluent method, which sets a proxy container.
StrimziKafkaContainer	withSaslPassword(String saslPassword)	Fluent method to set the SASL PLAIN mechanism's password.
StrimziKafkaContainer	withSaslUsername(String saslUsername)	Fluent method to set the SASL PLAIN mechanism's username.

Methods inherited from class org.testcontainers.containers.GenericContainer

addEnv, addExposedPort, addExposedPorts, addFileSystemBind, addLink, copyFileFromContainer, dependsOn, dependsOn, dependsOn, equals, getBinds, getCommandParts, getContainerId, getContainerInfo, getContainerName, getCopyToFileContainerPathMap, getCreateContainerCmdModifiers, getDependencies, getDockerClient, getDockerImageName, getEnv, getEnvMap, getExposedPorts, getExtraHosts, getImage, getIpAddress, getLabels, getLinkedContainers, getLivenessCheckPortNumbers, getLogConsumers, getNetwork, getNetworkAliases, getNetworkMode, getPortBindings, getShmSize, getStartupAttempts, getStartupCheckStrategy, getTestHostIpAddress, getTmpFsMapping, getVolumesFroms, getWorkingDirectory, hashCode, isHostAccessible, isPrivilegedMode, isShouldBeReused, setBinds, setCommand, setCommand, setCommandParts, setContainerDef, setCopyToFileContainerPathMap, setDockerImageName, setEnv, setExposedPorts, setExtraHosts, setHostAccessible, setImage, setLabels, setLinkedContainers, setLogConsumers, setNetwork, setNetworkAliases, setNetworkMode, setPortBindings, setPrivilegedMode, setShmSize, setStartupAttempts, setStartupCheckStrategy, setTmpFsMapping, setVolumesFroms, setWaitStrategy, setWorkingDirectory, start, toString, waitingFor, withAccessToHost, withClasspathResourceMapping, withClasspathResourceMapping, withCommand, withCommand, withCopyFileToContainer, withCopyToContainer, withCreateContainerCmdModifier, withEnv, withEnv, withExposedPorts, withExtraHost, withFileSystemBind, withImagePullPolicy, withLabel, withLabels, withLogConsumer, withMinimumRunningDuration, withNetwork, withNetworkAliases, withNetworkMode, withPrivilegedMode, withReuse, withSharedMemorySize, withStartupAttempts, withStartupCheckStrategy, withStartupTimeout, withTmpFs, withVolumesFrom, withWorkingDirectory

Methods inherited from class java.lang.Object

getClass, notify, notifyAll, wait, wait, wait

Methods inherited from interface org.testcontainers.containers.Container

addFileSystemBind, followOutput, followOutput, self, withEnv, withFileSystemBind

Methods inherited from interface org.testcontainers.containers.ContainerState

copyFileFromContainer, copyFileToContainer, copyFileToContainer, execInContainer, execInContainer, execInContainer, execInContainer, execInContainerWithUser, execInContainerWithUser, getBoundPortNumbers, getContainerIpAddress, getCurrentContainerInfo, getFirstMappedPort, getHost, getLogs, getLogs, getMappedPort, isCreated, isHealthy, isRunning

Methods inherited from interface org.testcontainers.lifecycle.Startable

close, getDependencies, start

Field Details

STARTER_SCRIPT

public static final String STARTER_SCRIPT

The file containing the startup script.

See Also:

• Constant Field Values

KAFKA_PORT

public static final int KAFKA_PORT

Default Kafka port

See Also:

• Constant Field Values

CONTROLLER_PORT

public static final int CONTROLLER_PORT

Default Kafka controller port

See Also:

• Constant Field Values

Method Details

stop

public void stop()

Specified by:

stop in interface org.testcontainers.lifecycle.Startable

Overrides:

stop in class org.testcontainers.containers.GenericContainer<StrimziKafkaContainer>

waitForRunning

public StrimziKafkaContainer waitForRunning()

Fluent method, which sets a waiting strategy to wait until the node is ready.

This method waits for a log message in the node log. The specific message depends on the node role: - Broker and combined-role nodes wait for "Transitioning from RECOVERY to RUNNING" - Controller-only nodes wait for "Kafka Server started" You can customize the strategy using GenericContainer.waitingFor(WaitStrategy).

Returns:

StrimziKafkaContainer instance

getBootstrapServers

public String getBootstrapServers()

Retrieves the bootstrap servers URL for Kafka clients.

Specified by:

getBootstrapServers in interface KafkaContainer

Returns:

the bootstrap servers URL

getNetworkBootstrapServers

public String getNetworkBootstrapServers()

Get the bootstrap servers that containers on the same network should use to connect

Returns:

a comma separated list of Kafka bootstrap servers

getBootstrapControllers

public String getBootstrapControllers()

Retrieves the bootstrap controllers URL for admin clients that need to connect to controllers. This is required for certain admin operations in KRaft mode, such as describing the cluster or performing controller-specific operations.

Specified by:

getBootstrapControllers in interface KafkaContainer

Returns:

the bootstrap controllers URL

Throws:

UnsupportedOperationException - if this node doesn't have controller role

getNetworkBootstrapControllers

public String getNetworkBootstrapControllers()

Get the bootstrap controllers that containers on the same network should use to connect to controllers

Returns:

a comma separated list of Kafka controller endpoints

Throws:

UnsupportedOperationException - if this node doesn't have controller role

getClusterId

public String getClusterId()

Get the cluster id. This is only supported for KRaft containers.

Returns:

The cluster id.

withKafkaConfigurationMap

public StrimziKafkaContainer withKafkaConfigurationMap(Map<String,String> kafkaConfigurationMap)

Fluent method, which sets kafkaConfigurationMap.

Parameters:

kafkaConfigurationMap - kafka configuration

Returns:

StrimziKafkaContainer instance

withNodeId

public StrimziKafkaContainer withNodeId(int nodeId)

Fluent method that sets the node ID.

Parameters:

nodeId - the node ID

Returns:

StrimziKafkaContainer instance

withKafkaVersion

public StrimziKafkaContainer withKafkaVersion(String kafkaVersion)

Fluent method, which sets kafkaVersion.

Parameters:

kafkaVersion - kafka version

Returns:

StrimziKafkaContainer instance

withPort

public StrimziKafkaContainer withPort(int fixedPort)

Fluent method, which sets fixed exposed port.

Parameters:

fixedPort - fixed port to expose

Returns:

StrimziKafkaContainer instance

withBootstrapServers

public StrimziKafkaContainer withBootstrapServers(Function<StrimziKafkaContainer,String> provider)

Fluent method, assign provider for overriding bootstrap servers string

Parameters:

provider - provider function for bootstrapServers string

Returns:

StrimziKafkaContainer instance

withProxyContainer

public StrimziKafkaContainer withProxyContainer(org.testcontainers.toxiproxy.ToxiproxyContainer proxyContainer)

Fluent method, which sets a proxy container. This container allows to create a TCP proxy between test code and Kafka broker. Every Kafka broker request will pass through the proxy where you can simulate network conditions (i.e. connection cut, latency).

Parameters:

proxyContainer - Proxy container

Returns:

StrimziKafkaContainer instance

withOAuthConfig

public StrimziKafkaContainer withOAuthConfig(String realm,
 String clientId,
 String clientSecret,
 String oauthUri,
 String usernameClaim)

Fluent method to configure OAuth settings.

Parameters:

realm - the realm

clientId - the OAuth client ID

clientSecret - the OAuth client secret

oauthUri - the OAuth URI

usernameClaim - the preferred username claim for OAuth

Returns:

StrimziKafkaContainer instance

withAuthenticationType

public StrimziKafkaContainer withAuthenticationType(AuthenticationType authType)

Sets the authentication type for the Kafka container.

Parameters:

authType - The authentication type to enable.

Returns:

StrimziKafkaContainer instance for method chaining.

withSaslUsername

public StrimziKafkaContainer withSaslUsername(String saslUsername)

Fluent method to set the SASL PLAIN mechanism's username.

Parameters:

saslUsername - The desired SASL username.

Returns:

StrimziKafkaContainer instance for method chaining.

withSaslPassword

public StrimziKafkaContainer withSaslPassword(String saslPassword)

Fluent method to set the SASL PLAIN mechanism's password.

Parameters:

saslPassword - The desired SASL password.

Returns:

StrimziKafkaContainer instance for method chaining.

withNodeRole

public StrimziKafkaContainer withNodeRole(KafkaNodeRole nodeRole)

Fluent method to set the Kafka node role.

Parameters:

nodeRole - the role this node should play in the cluster

Returns:

StrimziKafkaContainer instance for method chaining

withKafkaLog

public StrimziKafkaContainer withKafkaLog(org.apache.logging.log4j.Level level)

Configures the Kafka container to use the specified logging level for Kafka logs and the io.strimzi logger.

This method generates a custom log4j2.yaml file with the desired logging level and copies it into the Kafka container. By setting the logging level, you can control the verbosity of Kafka's log output, which is useful for debugging or monitoring purposes.

Example Usage:

 StrimziKafkaContainer kafkaContainer = new StrimziKafkaContainer()
     .withKafkaLog(Level.DEBUG)
     .start();
 

Parameters:

level - the desired Level of logging (e.g., DEBUG, INFO, WARN, ERROR)

Returns:

the current instance of StrimziKafkaContainer for method chaining

withLogCollection

public StrimziKafkaContainer withLogCollection()

Fluent method to enable log collection with a default log file path. The actual filename is determined at runtime when logs are collected.

Returns:

StrimziKafkaContainer instance for method chaining

withLogCollection

public StrimziKafkaContainer withLogCollection(String logFilePath)

Fluent method to enable log collection for all containers in the cluster with a custom log file path. If the path ends with "/", role-based filenames are automatically appended at runtime for each container: Controller-only: "kafka-controller-{nodeId}.log" Broker-only: "kafka-broker-{nodeId}.log" Combined: "kafka-container-{nodeId}.log" otherwise, the path doesn't end with "/", it's used as the exact filename base for all containers

Parameters:

logFilePath - the base path where container logs will be saved. Use "/" suffix for automatic role-based naming.

Returns:

the current instance of StrimziKafkaClusterBuilder for method chaining

getProxy

public eu.rekawek.toxiproxy.Proxy getProxy()

Retrieves the Proxy instance for this Kafka broker.

Returns:

Proxy instance for this Kafka broker.

getNodeId

public int getNodeId()

Gets the node ID for this Kafka container.

Returns:

the node ID

isOAuthEnabled

public boolean isOAuthEnabled()

Checks if OAuth is enabled.

Returns:

true if OAuth is enabled; false otherwise

getSaslUsername

public String getSaslUsername()

Gets the SASL username for authentication.

Returns:

the SASL username

getSaslPassword

public String getSaslPassword()

Gets the SASL password for authentication.

Returns:

the SASL password

getRealm

public String getRealm()

Gets the OAuth realm.

Returns:

the OAuth realm

getClientId

public String getClientId()

Gets the OAuth client ID.

Returns:

the OAuth client ID

getClientSecret

public String getClientSecret()

Gets the OAuth client secret.

Returns:

the OAuth client secret

getOauthUri

public String getOauthUri()

Gets the OAuth URI.

Returns:

the OAuth URI

getUsernameClaim

public String getUsernameClaim()

Gets the OAuth username claim.

Returns:

the OAuth username claim

getAuthenticationType

public AuthenticationType getAuthenticationType()

Gets the authentication type configured for this container.

Returns:

the authentication type

getNodeRole

public KafkaNodeRole getNodeRole()

Gets the node role for this Kafka container.

Returns:

the node role