HosebirdReader { @SuppressWarnings("checkstyle:designforextension") protected Producer<String, String> getKafkaProducer( final KafkaConfig config) { log.info("Creating Kafka producer"); Properties properties = new Properties(); properties.put( ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, config.servers ); properties.put( ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); properties.put( ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); properties.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, "1000"); properties.put(ProducerConfig.RECONNECT_BACKOFF_MS_CONFIG, "1000"); return new KafkaProducer<>(properties); } HosebirdReader(); final void run(final String[] args);  }
@Test public void getkafkaproducer_should_create_producer_correctly() { HosebirdReader tr = new ConcreteHosebirdReader(); Producer<String, String> producer = tr.getKafkaProducer(new KafkaConfig()); assertNotNull(producer); }
@Test public void should_set_kafka_config_on_producer() { HosebirdReader tr = new ConcreteHosebirdReader(); KafkaConfig config = new KafkaConfig(); try { Producer<String, String> producerWithConfig = tr.getKafkaProducer(config); Field f1 = producerWithConfig.getClass().getDeclaredField("producerConfig"); f1.setAccessible(true); ProducerConfig pc = (ProducerConfig)f1.get(producerWithConfig); assertEquals(1000, pc.getLong(ProducerConfig.RETRY_BACKOFF_MS_CONFIG)); Field f2 = producerWithConfig.getClass().getDeclaredField("producerConfig"); f2.setAccessible(true); ProducerConfig pc2 = (ProducerConfig)f2.get(producerWithConfig); assertEquals(1000, pc2.getLong(ProducerConfig.RECONNECT_BACKOFF_MS_CONFIG)); } catch (Exception e) { assertNull(e); } }