KafkaProducer implements Producer { @Override public void send(final Message message, final TestContext context) { Assert.notNull(message, "Message is empty - unable to send empty message"); String topic = Optional.ofNullable(message.getHeader(KafkaMessageHeaders.TOPIC)) .map(Object::toString) .map(context::replaceDynamicContentInString) .orElse(context.replaceDynamicContentInString(endpointConfiguration.getTopic())); if (!StringUtils.hasText(topic)) { throw new CitrusRuntimeException(String.format("Invalid Kafka stream topic header %s - must not be empty or null", KafkaMessageHeaders.TOPIC)); } if (log.isDebugEnabled()) { log.debug("Sending Kafka stream message to topic: '" + topic + "'"); } try { ProducerRecord<Object, Object> producerRecord = endpointConfiguration.getMessageConverter().convertOutbound(message, endpointConfiguration, context); producer.send(producerRecord).get(endpointConfiguration.getTimeout(), TimeUnit.MILLISECONDS); log.info("Message was sent to Kafka stream topic: '" + topic + "'"); } catch (InterruptedException | ExecutionException e) { throw new CitrusRuntimeException(String.format("Failed to send message to Kafka topic '%s'", topic), e); } catch (TimeoutException e) { throw new CitrusRuntimeException(String.format("Failed to send message to Kafka topic '%s' - timeout after %s milliseconds", topic, endpointConfiguration.getTimeout()), e); } context.onOutboundMessage(message); } KafkaProducer(String name, KafkaEndpointConfiguration endpointConfiguration); @Override void send(final Message message, final TestContext context); @Override String getName(); void setProducer(org.apache.kafka.clients.producer.KafkaProducer<Object, Object> producer);  }
@Test public void testSendMessageTimeout() { KafkaEndpoint endpoint = new KafkaEndpoint(); endpoint.getEndpointConfiguration().setServer("localhost:" + SocketUtils.findAvailableTcpPort()); endpoint.getEndpointConfiguration().setTopic("test"); endpoint.getEndpointConfiguration().setProducerProperties(Collections.singletonMap(ProducerConfig.MAX_BLOCK_MS_CONFIG, 1000)); try { endpoint.createProducer().send(new KafkaMessage("foo"), context); } catch(CitrusRuntimeException e) { Assert.assertEquals(e.getMessage(), "Failed to send message to Kafka topic 'test'"); Assert.assertEquals(e.getCause().getClass(), java.util.concurrent.ExecutionException.class); Assert.assertTrue(e.getCause().getMessage().contains("org.apache.kafka.common.errors.TimeoutException")); return; } Assert.fail("Missing " + CitrusRuntimeException.class + " because of message timeout"); }
@Test public void testSendEmptyMessage() { KafkaEndpoint endpoint = new KafkaEndpoint(); endpoint.getEndpointConfiguration().setServer("localhost:9092"); endpoint.getEndpointConfiguration().setTopic("test"); try { endpoint.createProducer().send(null, context); } catch(IllegalArgumentException e) { Assert.assertEquals(e.getMessage(), "Message is empty - unable to send empty message"); return; } Assert.fail("Missing " + CitrusRuntimeException.class + " because of sending empty message"); }