public static KafkaWrapper create ( State state ) { Preconditions . checkNotNull ( state . getProp ( ConfigurationKeys . KAFKA_BROKERS ) , "Need to specify at least one Kafka broker." ) ; KafkaWrapper . Builder builder = new KafkaWrapper . Builder ( ) ; if ( state . getPropAsBoolean ( USE_NEW_KAFKA_API , DEFAULT_USE_NEW_KAFKA_API ) ) { builder = builder . withNewKafkaAPI ( ) ; } Config config = ConfigUtils . propertiesToConfig ( state . getProperties ( ) ) ; return builder . withBrokers ( state . getPropAsList ( ConfigurationKeys . KAFKA_BROKERS ) ) . withConfig ( config ) . build ( ) ; }
Create a KafkaWrapper based on the given type of Kafka API and list of Kafka brokers .