GridMeasurementPublishedEventProducer { public void send(final String measurement) { final GridMeasurementPublishedEvent event = this.mapper.map(measurement, GridMeasurementPublishedEvent.class); this.kafkaTemplate.sendDefault(event); } @Autowired  GridMeasurementPublishedEventProducer(
            @Qualifier("distributionAutomationKafkaTemplate") final KafkaTemplate<String, GridMeasurementPublishedEvent> kafkaTemplate,
            final DistributionAutomationMapper mapper); void send(final String measurement);  }
@Test void sendTest() { this.producer.send("TST-01; 220.1; 220.2; 220.3; 5.1; 5.2; 5.3; 7.1; 7.2; 7.3;"); final Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("testGroup", "true", this.embeddedKafka); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); final ConsumerFactory<String, GridMeasurementPublishedEvent> consumerFactory = new DefaultKafkaConsumerFactory<>( consumerProps, new StringDeserializer(), new GridMeasurementPublishedEventDeserializer()); final Consumer<String, GridMeasurementPublishedEvent> consumer = consumerFactory.createConsumer(); this.embeddedKafka.consumeFromAnEmbeddedTopic(consumer, this.topic); final ConsumerRecord<String, GridMeasurementPublishedEvent> received = KafkaTestUtils.getSingleRecord(consumer, this.topic); assertThat(received).has(value(this.message)); }