@Test public void testTimeoutConfig() { String brokerList = "localhost:9092"; Properties props = new Properties(); props.setProperty(ConfigurationKeys.KAFKA_BROKERS, brokerList); props.setProperty("source.kafka.fetchTimeoutMillis", "10000"); props.setProperty("source.kafka.socketTimeoutMillis", "1000"); State state = new State(props); try { KafkaWrapper wrapper = KafkaWrapper.create(state); Assert.fail("KafkaWrapper should fail to initialize if fetchTimeout is greater than socketTimeout"); } catch (IllegalArgumentException e) { log.info("Found exception as expected"); log.debug("Exception trace", e); } catch (Exception e) { Assert.fail("Should only throw IllegalArgumentException", e); } }
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(); }
KafkaWrapper implements Closeable { 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(); } }
KafkaWrapper implements Closeable { 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(); } private  KafkaWrapper(Builder builder); }
KafkaWrapper implements Closeable { 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(); } private  KafkaWrapper(Builder builder); static KafkaWrapper create(State state); List<String> getBrokers(); List<KafkaTopic> getFilteredTopics(List<Pattern> blacklist, List<Pattern> whitelist); long getEarliestOffset(KafkaPartition partition); long getLatestOffset(KafkaPartition partition); Iterator<MessageAndOffset> fetchNextMessageBuffer(KafkaPartition partition, long nextOffset, long maxOffset); @Override void close(); }
KafkaWrapper implements Closeable { 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(); } private  KafkaWrapper(Builder builder); static KafkaWrapper create(State state); List<String> getBrokers(); List<KafkaTopic> getFilteredTopics(List<Pattern> blacklist, List<Pattern> whitelist); long getEarliestOffset(KafkaPartition partition); long getLatestOffset(KafkaPartition partition); Iterator<MessageAndOffset> fetchNextMessageBuffer(KafkaPartition partition, long nextOffset, long maxOffset); @Override void close();  }