KafkaInput implements StreamInput, CanRecordProgress, ProvidesAlias,
    ProvidesValidations, InstantiatesComponents, UsesExpectedSchema { @Override public void configure(Config config) { this.config = config; groupID = ConfigUtils.getOrElse(config, GROUP_ID_CONFIG, UUID.randomUUID().toString()); kafkaParams = Maps.newHashMap(); kafkaParams.put("bootstrap.servers", config.getString(BROKERS_CONFIG)); kafkaParams.put("group.id", groupID); kafkaParams.put("enable.auto.commit", "false"); KafkaCommon.addCustomParams(kafkaParams, config); topics = Sets.newHashSet(config.getStringList(TOPICS_CONFIG)); if (ConfigUtils.getOrElse(config, WINDOW_ENABLED_CONFIG, false) && doesRecordProgress(config)) { throw new RuntimeException("Kafka input offset management not compatible with stream windowing."); } }  @Override void configure(Config config); @Override void receiveExpectedSchema(StructType expectedSchema); @Override JavaDStream<?> getDStream(); @Override Function<?, Row> getMessageEncoderFunction(); @Override StructType getProvidingSchema(); @Override String getAlias(); @Override void recordProgress(JavaRDD<?> batch); @Override Validations getValidations(); @Override Set<InstantiatedComponent> getComponents(Config config, boolean configure); static final String BROKERS_CONFIG; static final String TOPICS_CONFIG; static final String WINDOW_ENABLED_CONFIG; static final String WINDOW_MILLISECONDS_CONFIG; static final String WINDOW_SLIDE_MILLISECONDS_CONFIG; static final String OFFSETS_MANAGE_CONFIG; static final String OFFSETS_OUTPUT_CONFIG; static final String GROUP_ID_CONFIG; }
@Test public void testMultipleTopicsConfigure() { KafkaInput kafkaInput = new KafkaInput(); new Expectations() { { config.getStringList(KafkaInput.TOPICS_CONFIG); returns(Lists.newArrayList("foo", "bar", "bar")); } }; kafkaInput.configure(config); assertEquals(kafkaInput.topics.size(), 2); assertTrue(kafkaInput.topics.contains("foo")); assertTrue(kafkaInput.topics.contains("bar")); }