Использование встроенного Kafka в весеннем тесте облачного потока с пользовательскими привязками каналов

Question

У меня есть приложение весенней загрузки, в котором я использую spring-cloud-stream для использования темы кафки, некоторой обработки и публикации в другой теме кафки. Приложение работает нормально, и я написал модульные тесты (с использованием TestBinder), которые также работают нормально.

Сейчас я пытаюсь написать интеграционный тест со встроенным Kafka и протестировать сквозное тестирование. функциональность. Я следовал примеру здесь https://github.com/spring-cloud/spring-cloud-stream-samples/blob/master/testing-samples/test-embedded-kafka/src/test/java/demo/EmbeddedKafkaApplicationTests.java, чтобы написать тест, однако это не работает - я не могу получить никакого сообщения по теме вывода.

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());
    }

}

Приведенный выше тест не пройден, поскольку я ожидаю присутствия 1 записи, но получаю 0 записей в теме вывода.

Answer 1

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

Необходимо дождаться подписки;0 не сделает этого;образец ожидает до 10 секунд.

Однако безопаснее использовать

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

, поскольку он надежно ожидает назначения с использованием ConsumerRebalanceListener.

После подписки вы также можете использовать

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

, чтобы получить запись (если вы ожидаете только одну, или getRecords(...) в противном случае).