public void startup ( ) { if ( config . getEnableZookeeper ( ) ) { serverRegister . registerBrokerInZk ( ) ; for ( String topic : getAllTopics ( ) ) { serverRegister . processTask ( new TopicTask ( TopicTask . TaskType . CREATE , topic ) ) ; } startupLatch . countDown ( ) ; } logger . debug ( "Starting log flusher every {} ms with the following overrides {}" , config . getFlushSchedulerThreadRate ( ) , logFlushIntervalMap ) ; logFlusherScheduler . scheduleWithRate ( new Runnable ( ) { public void run ( ) { flushAllLogs ( false ) ; } } , config . getFlushSchedulerThreadRate ( ) , config . getFlushSchedulerThreadRate ( ) ) ; }
Register this broker in ZK for the first time .