KafkaProducerManager { public Producer getProducer(KafkaPublishTask.Input input) { Properties configProperties = getProducerProperties(input); return getFromCache(configProperties, () -> new KafkaProducer(configProperties)); } KafkaProducerManager(Configuration configuration); Producer getProducer(KafkaPublishTask.Input input); static final String KAFKA_PUBLISH_REQUEST_TIMEOUT_MS; static final String STRING_SERIALIZER; static final String DEFAULT_REQUEST_TIMEOUT; static final RemovalListener<Properties, Producer> LISTENER; final String requestTimeoutConfig; }
@Test(expected = RuntimeException.class) public void testExecutionException() { Configuration configuration = getConfiguration(); KafkaProducerManager manager = new KafkaProducerManager(configuration); KafkaPublishTask.Input input = getInput(); Producer producer = manager.getProducer(input); Assert.assertNotNull(producer); }