KafkaProducerManager { @VisibleForTesting Properties getProducerProperties(KafkaPublishTask.Input input) { Properties configProperties = new Properties(); configProperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, input.getBootStrapServers()); configProperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, input.getKeySerializer()); String requestTimeoutMs = requestTimeoutConfig; if (Objects.nonNull(input.getRequestTimeoutMs())) { requestTimeoutMs = String.valueOf(input.getRequestTimeoutMs()); } String maxBlockMs = maxBlockMsConfig; if (Objects.nonNull(input.getMaxBlockMs())) { maxBlockMs = String.valueOf(input.getMaxBlockMs()); } configProperties.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, requestTimeoutMs); configProperties.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, maxBlockMs); configProperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, STRING_SERIALIZER); return 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 public void testRequestTimeoutSetFromDefault() { KafkaProducerManager manager = new KafkaProducerManager(new SystemPropertiesConfiguration()); KafkaPublishTask.Input input = getInput(); Properties props = manager.getProducerProperties(input); Assert.assertEquals(props.getProperty(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG), "100"); }
@Test public void testRequestTimeoutSetFromInput() { KafkaProducerManager manager = new KafkaProducerManager(new SystemPropertiesConfiguration()); KafkaPublishTask.Input input = getInput(); input.setRequestTimeoutMs(200); Properties props = manager.getProducerProperties(input); Assert.assertEquals(props.getProperty(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG), "200"); }
@Test public void testRequestTimeoutSetFromConfig() { Configuration configuration = getConfiguration(); KafkaProducerManager manager = new KafkaProducerManager(configuration); KafkaPublishTask.Input input = getInput(); Properties props = manager.getProducerProperties(input); Assert.assertEquals(props.getProperty(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG), "150"); }
@Test public void testMaxBlockMsFromConfig() { Configuration configuration = getConfiguration(); KafkaProducerManager manager = new KafkaProducerManager(configuration); KafkaPublishTask.Input input = getInput(); Properties props = manager.getProducerProperties(input); Assert.assertEquals(props.getProperty(ProducerConfig.MAX_BLOCK_MS_CONFIG), "500"); }
@Test public void testMaxBlockMsFromInput() { Configuration configuration = getConfiguration(); KafkaProducerManager manager = new KafkaProducerManager(configuration); KafkaPublishTask.Input input = getInput(); input.setMaxBlockMs(600); Properties props = manager.getProducerProperties(input); Assert.assertEquals(props.getProperty(ProducerConfig.MAX_BLOCK_MS_CONFIG), "600"); }