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 ) ) ; }
Start embedded server instances for Kafka and Zookeeper .