code: 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();
    }
docs: Create a KafkaWrapper based on the given type of Kafka API and list of Kafka brokers.
  
  @param state A {@link State} object that should contain a list of comma separated Kafka brokers
  in property "kafka.brokers". It may optionally specify whether to use the new Kafka API by setting
  use.new.kafka.api=true.