Using embedded Kafka in spring cloud stream test with custom channel bindings

I have a spring boot application where I am using spring-cloud-stream to consume from a kafka topic, do some processing and publish to another kafka topic. The application works fine and I've written unit tests (using the TestBinder) which are running fine as well.

I am now trying to write an integration test with an embedded Kafka and test the end-to-end functionality. I have followed the sample here https://github.com/spring-cloud/spring-cloud-stream-samples/blob/master/testing-samples/test-embedded-kafka/src/test/java/demo/EmbeddedKafkaApplicationTests.java to write the test however this is not working - I am unable to receive any message on the output topic.

application.yml

spring:
  cloud:
    stream:
      bindings:
        incoming-message:
          destination: ReadyForProcessing
          content-type: application/json
          group: ReadyForProcessingGroup
        outgoing-message:
          destination: TransactionSettled
          content-type: application/json

TransformerBinding.java

public interface TransformerBinding {

    String INCOMING_MESSAGE = "incoming-message";

    String OUTGOING_MESSAGE = "outgoing-message";

    @Input(INCOMING_MESSAGE)
    SubscribableChannel incomingMessage();

    @Output(OUTGOING_MESSAGE)
    MessageChannel outgoingMessage();

}

EventProcessor.java

@Service
@EnableBinding(TransformerBinding.class)
@Slf4j
@AllArgsConstructor
public class EventProcessor {

    @Transformer(inputChannel = TransformerBinding.INCOMING_MESSAGE, outputChannel = TransformerBinding.OUTGOING_MESSAGE)
    public TransactionSettledEvent transform(@Payload final ReadyForProcessingEvent readyForProcessingEvent) {
        log.info("Event received in processor: {}", readyForProcessingEvent);


        return TransactionSettledEvent.builder().transactionRef(readyForProcessingEvent.getTransactionRef()).status("Settled").build();
    }

}

EventProcessorTest.java

@RunWith(SpringRunner.class)
@SpringBootTest(properties = "spring.autoconfigure.exclude="
        + "org.springframework.cloud.stream.test.binder.TestSupportBinderAutoConfiguration")
public class EventProcessorIT {

    private static final String INPUT_TOPIC = "ReadyForProcessing";
    private static final String OUTPUT_TOPIC = "TransactionSettled";
    private static final String CONSUMER_GROUP = "TestConsumerGroup";

    @Autowired
    private ObjectMapper mapper;

    @ClassRule
    public static EmbeddedKafkaRule embeddedKafka = new EmbeddedKafkaRule(1, true, INPUT_TOPIC, OUTPUT_TOPIC);

    @BeforeClass
    public static void setup() {
        System.setProperty("spring.cloud.stream.kafka.binder.brokers", embeddedKafka.getEmbeddedKafka().getBrokersAsString());
    }

    @Test
    public void testSendReceive() {
        Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka.getEmbeddedKafka());
        senderProps.put("key.serializer", StringSerializer.class);
        senderProps.put("value.serializer", JsonSerializer.class);
        DefaultKafkaProducerFactory<String, ReadyForProcessingEvent> pf = new DefaultKafkaProducerFactory<>(senderProps);
        KafkaTemplate<String, ReadyForProcessingEvent> template = new KafkaTemplate<>(pf, true);
        template.setDefaultTopic(INPUT_TOPIC);
        template.sendDefault(ReadyForProcessingEvent.builder().transactionRef("123456").build());

        Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(CONSUMER_GROUP, "false", embeddedKafka.getEmbeddedKafka());
        consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, CONSUMER_GROUP);
        consumerProps.put("key.deserializer", StringDeserializer.class);
        consumerProps.put("value.deserializer", JsonDeserializer.class);
        DefaultKafkaConsumerFactory<String, TransactionSettledEvent> cf = new DefaultKafkaConsumerFactory<>(consumerProps);

        Consumer<String, TransactionSettledEvent> consumer = cf.createConsumer();
        consumer.subscribe(Collections.singleton(OUTPUT_TOPIC));
        ConsumerRecords<String, TransactionSettledEvent> records = consumer.poll(0);
        consumer.commitSync();

        assertEquals("Only 1 record should be received as response", 1, records.count());
        final TransactionSettledEvent transactionSettledEvent = this.mapper.convertValue(records.iterator().next().value(), TransactionSettledEvent.class);
        assertEquals("Output event not as expected", "Settled", transactionSettledEvent.getStatus());
    }

}

The test above fails because I'm expecting 1 record to be present but I'm getting 0 record at the output topic.

ConsumerRecords<String, TransactionSettledEvent> records = consumer.poll(0);

You need wait for the subscription to occur; 0 won't do it; the sample waits for up to 10 seconds.

However, it's safer to use

embeddedKafkaRule().getEmbeddedKafka().consumeFromAnEmbeddedTopic(...);

because it reliably waits for assignment using a ConsumerRebalanceListener.

Once subscribed, you can also use

KafkaTestUtils.getSingleRecord(Consumer<K, V> consumer, String topic);

to get the record (if you only expect one, or getRecords(...) otherwise).