2

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でテストを記述しますが、これは機能しません。出力トピックに関するメッセージを受け取ることができません。

アプリケーション.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 レコードを取得しているため、失敗します。

4

1 に答える 1

1

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

サブスクリプションが発生するまで待つ必要があります。0 ではできません。サンプルは最大 10 秒間待機します。

ただし、使用する方が安全です

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

を使用して割り当てを確実に待機するためConsumerRebalanceListenerです。

登録すると、次も使用できます

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

レコードを取得します (1 つだけを期待する場合、またはgetRecords(...)そうでない場合)。

于 2019-10-12T13:24:44.810 に答える