code: public void start() {
          if (kafkaServer != null) {
              log.warn("Found instance of Kafka server - avoid duplicate Kafka server startup");
              return;
          }
  
          File logDir = createLogDir();
          zookeeper = createZookeeperServer(logDir);
          serverFactory = createServerFactory();
  
          try {
              serverFactory.startup(zookeeper);
          } catch (InterruptedException | IOException e) {
              throw new CitrusRuntimeException("Failed to start embedded zookeeper server", e);
          }
  
          Properties brokerConfigProperties = createBrokerProperties("localhost:" + zookeeperPort, kafkaServerPort, logDir);
          brokerConfigProperties.setProperty(KafkaConfig.ReplicaSocketTimeoutMsProp(), "1000");
          brokerConfigProperties.setProperty(KafkaConfig.ControllerSocketTimeoutMsProp(), "1000");
          brokerConfigProperties.setProperty(KafkaConfig.OffsetsTopicReplicationFactorProp(), "1");
          brokerConfigProperties.setProperty(KafkaConfig.ReplicaHighWatermarkCheckpointIntervalMsProp(), String.valueOf(Long.MAX_VALUE));
  
          if (brokerProperties != null) {
              brokerProperties.forEach(brokerConfigProperties::put);
          }
  
          kafkaServer = new KafkaServer(new KafkaConfig(brokerConfigProperties), Time.SYSTEM,
                  scala.Option.apply(null), scala.collection.JavaConversions.asScalaBuffer(Collections.<KafkaMetricsReporter>emptyList()).toList());
          kafkaServer.startup();
          kafkaServer.boundPort(ListenerName.forSecurityProtocol(SecurityProtocol.PLAINTEXT));
  
          createKafkaTopics(StringUtils.commaDelimitedListToSet(topics));
      }
docs: Start embedded server instances for Kafka and Zookeeper.