diff --git a/src/main/java/io/zeebe/containers/cluster/ZeebeClusterBuilder.java b/src/main/java/io/zeebe/containers/cluster/ZeebeClusterBuilder.java
index 9d9def77..2f141ca4 100644
--- a/src/main/java/io/zeebe/containers/cluster/ZeebeClusterBuilder.java
+++ b/src/main/java/io/zeebe/containers/cluster/ZeebeClusterBuilder.java
@@ -18,6 +18,7 @@
 import io.zeebe.containers.ZeebeBrokerContainer;
 import io.zeebe.containers.ZeebeBrokerNode;
 import io.zeebe.containers.ZeebeContainer;
+import io.zeebe.containers.ZeebeDefaults;
 import io.zeebe.containers.ZeebeGatewayContainer;
 import io.zeebe.containers.ZeebeGatewayNode;
 import io.zeebe.containers.ZeebeTopologyWaitStrategy;
@@ -31,6 +32,7 @@
 import org.apiguardian.api.API.Status;
 import org.testcontainers.containers.GenericContainer;
 import org.testcontainers.containers.Network;
+import org.testcontainers.utility.DockerImageName;
 
 /**
  * Convenience class to help build Zeebe clusters.
@@ -94,6 +96,7 @@ public class ZeebeClusterBuilder {
   private static final String BROKER_NETWORK_ALIAS_PREFIX = "zeebe-broker-";
   private static final String GATEWAY_NETWORK_ALIAS_PREFIX = "zeebe-gateway-";
   private static final String DEFAULT_CLUSTER_NAME = "zeebe-cluster";
+  private static final ZeebeDefaults ZEEBE_DEFAULTS = ZeebeDefaults.getInstance();
 
   private Network network = Network.SHARED;
   private String name = DEFAULT_CLUSTER_NAME;
@@ -104,6 +107,8 @@ public class ZeebeClusterBuilder {
   private int partitionsCount = 1;
   private int replicationFactor = 1;
   private boolean useEmbeddedGateway = true;
+  private DockerImageName gatewayImageName = ZEEBE_DEFAULTS.getDefaultDockerImage();
+  private DockerImageName brokerImageName = ZEEBE_DEFAULTS.getDefaultDockerImage();
 
   private final Map<String, ZeebeGatewayNode<? extends GenericContainer<?>>> gateways =
       new HashMap<>();
@@ -242,6 +247,41 @@ public ZeebeClusterBuilder withName(final String name) {
     return this;
   }
 
+  /**
+   * Sets the docker image for separate gateways. This could be used to create a test against
+   * specific Zeebe version.
+   *
+   * @param gatewayImageName the docker image name of the Zeebe
+   * @return this builder instance for chaining
+   */
+  public ZeebeClusterBuilder withGatewayImage(final DockerImageName gatewayImageName) {
+    this.gatewayImageName = Objects.requireNonNull(gatewayImageName);
+    return this;
+  }
+
+  /**
+   * Sets the docker image for brokers with embedded gateways. This could be used to create a test
+   * against specific Zeebe version.
+   *
+   * @param brokerImageName the docker image name of the Zeebe
+   * @return this builder instance for chaining
+   */
+  public ZeebeClusterBuilder withBrokerImage(final DockerImageName brokerImageName) {
+    this.brokerImageName = Objects.requireNonNull(brokerImageName);
+    return this;
+  }
+
+  /**
+   * Sets the docker image for brokers with embedded gateways and separate gateways. This could be
+   * used to create a test against specific Zeebe version.
+   *
+   * @param imageName the docker image name of the Zeebe
+   * @return this builder instance for chaining
+   */
+  public ZeebeClusterBuilder withImage(final DockerImageName imageName) {
+    return this.withGatewayImage(imageName).withBrokerImage(imageName);
+  }
+
   /**
    * Builds a new Zeebe cluster. Will create {@link #brokersCount} brokers (accessible later via
    * {@link ZeebeCluster#getBrokers()}) and {@link #gatewaysCount} standalone gateways (accessible
@@ -305,13 +345,13 @@ private void createBrokers() {
       final ZeebeBrokerNode<?> broker;
 
       if (useEmbeddedGateway) {
-        final ZeebeContainer container = new ZeebeContainer();
+        final ZeebeContainer container = new ZeebeContainer(brokerImageName);
         configureGateway(container);
 
         broker = container;
         gateways.put(String.valueOf(i), container);
       } else {
-        broker = new ZeebeBrokerContainer();
+        broker = new ZeebeBrokerContainer(brokerImageName);
       }
 
       configureBroker(broker, i);
@@ -341,7 +381,7 @@ private void createStandaloneGateways() {
   }
 
   private ZeebeGatewayContainer createStandaloneGateway(final String memberId) {
-    final ZeebeGatewayContainer gateway = new ZeebeGatewayContainer();
+    final ZeebeGatewayContainer gateway = new ZeebeGatewayContainer(gatewayImageName);
 
     gateway
         .withNetwork(network)
diff --git a/src/test/java/io/zeebe/containers/cluster/ZeebeClusterBuilderTest.java b/src/test/java/io/zeebe/containers/cluster/ZeebeClusterBuilderTest.java
index 750d0f44..037b75bb 100644
--- a/src/test/java/io/zeebe/containers/cluster/ZeebeClusterBuilderTest.java
+++ b/src/test/java/io/zeebe/containers/cluster/ZeebeClusterBuilderTest.java
@@ -25,10 +25,12 @@
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
+import org.assertj.core.api.Condition;
 import org.junit.jupiter.api.Test;
 import org.testcontainers.containers.Container;
 import org.testcontainers.containers.GenericContainer;
 import org.testcontainers.containers.Network;
+import org.testcontainers.utility.DockerImageName;
 
 final class ZeebeClusterBuilderTest {
   @Test
@@ -506,4 +508,68 @@ void shouldNotConfigureEmbeddedGateway() {
         .as("the broker is not configured to enable the embedded gateway")
         .doesNotContainEntry("ZEEBE_BROKER_GATEWAY_ENABLE", "true");
   }
+
+  @Test
+  void shouldSetImageNameForGateways() {
+    // given
+    final ZeebeClusterBuilder builder = new ZeebeClusterBuilder();
+    final String zeebeDockerImage = "camunda/zeebe:latest";
+
+    // when
+    final DockerImageName gatewayImageName = DockerImageName.parse(zeebeDockerImage);
+    builder.withGatewayImage(gatewayImageName).withGatewaysCount(1).withEmbeddedGateway(false);
+    final ZeebeCluster cluster = builder.build();
+
+    // then
+    final Map<String, ZeebeGatewayNode<? extends GenericContainer<?>>> gateways =
+        cluster.getGateways();
+    assertThat(gateways).hasSize(1).hasValueSatisfying(zeebeImageHasImageName(zeebeDockerImage));
+  }
+
+  @Test
+  void shouldSetImageNameForBrokers() {
+    // given
+    final ZeebeClusterBuilder builder = new ZeebeClusterBuilder();
+    final String zeebeDockerImage = "camunda/zeebe:latest";
+
+    // when
+    final DockerImageName gatewayImageName = DockerImageName.parse(zeebeDockerImage);
+    builder.withBrokerImage(gatewayImageName).withBrokersCount(1);
+    final ZeebeCluster cluster = builder.build();
+
+    // then
+    final Map<Integer, ZeebeBrokerNode<? extends GenericContainer<?>>> brokers =
+        cluster.getBrokers();
+    assertThat(brokers).hasSize(1).hasValueSatisfying(zeebeImageHasImageName(zeebeDockerImage));
+  }
+
+  @Test
+  void shouldSetImageNameForGatewaysAndBrokers() {
+    // given
+    final ZeebeClusterBuilder builder = new ZeebeClusterBuilder();
+    final String zeebeDockerImage = "camunda/zeebe:latest";
+
+    // when
+    final DockerImageName gatewayImageName = DockerImageName.parse(zeebeDockerImage);
+    builder
+        .withImage(gatewayImageName)
+        .withBrokersCount(1)
+        .withGatewaysCount(1)
+        .withEmbeddedGateway(false);
+    final ZeebeCluster cluster = builder.build();
+
+    // then
+    final Map<Integer, ZeebeBrokerNode<? extends GenericContainer<?>>> brokers =
+        cluster.getBrokers();
+    assertThat(brokers).hasSize(1).hasValueSatisfying(zeebeImageHasImageName(zeebeDockerImage));
+    final Map<String, ZeebeGatewayNode<? extends GenericContainer<?>>> gateways =
+        cluster.getGateways();
+    assertThat(gateways).hasSize(1).hasValueSatisfying(zeebeImageHasImageName(zeebeDockerImage));
+  }
+
+  private Condition<ZeebeNode<? extends GenericContainer<?>>> zeebeImageHasImageName(
+      final String imageName) {
+    return new Condition<>(
+        node -> node.getDockerImageName().equals(imageName), "Image Name Condition");
+  }
 }