Сообщения тестирования контракта в Flux в Spring Cloud Stream дают исключение IllegalArgumentException, которое не должно быть нулевым

Question

Я пытаюсь заключить контракт на тестирование потока событий, генерируемых конечной точкой, с использованием Spring Cloud Stream и Spring Cloud Contract. Тем не менее, я получаю IllegalArgumentException с detailMessage Message must not be null при запуске моего теста. Или, другими словами, я не получаю никаких сообщений на моем приемнике сообщений. Моя весенняя облачная версия - Хокстон. Я добавил следующие зависимости в мой pom

    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-stream-binder-kafka</artifactId>
    </dependency>
    <!--TEST -->
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-stream-test-support</artifactId>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-contract-verifier</artifactId>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-stream</artifactId>
        <type>test-jar</type>
        <scope>test</scope>
        <classifier>test-binder</classifier>
    </dependency>

Я настроил новую привязку после примеров здесь . Работает правильно, когда я запускаю приложение и отправляю запрос curl -X POST localhost:8080.

application.properties

stubrunner.stream.enabled=true
spring.cloud.stream.bindings.uppercase-in-0.destination=test
spring.cloud.stream.bindings.uppercase-in-0.group=testGroup
spring.cloud.stream.bindings.uppercase-out-0.destination=hood
spring.cloud.stream.bindings.consume-in-0.destination=hood
spring.cloud.stream.bindings.consume-out-0.destination=downtown
spring.cloud.stream.bindings.supplier-out-0.destination=test
spring.cloud.function.definition=uppercase;consume;supplier

Контроллер

 @PostMapping(value = "/")
 public void handlePost() {
    Faker faker = new Faker();
    Message<String> message = MessageBuilder.withPayload(faker.chuckNorris().fact()).build();
    //    supplier.get();
    EventSupplier.processor.onNext(message);
 }

Определение компонента

@Configuration
public class EventSupplier {
  public static final EmitterProcessor<Message<String>> processor = EmitterProcessor.create();

  @Bean
  //  public Supplier<String> supplier() {
  public Supplier<Flux<Message<String>>> supplier() {
    //    return () -> "";
    return () -> processor;
  }

}

Контракты Базовый тест

@ActiveProfiles("test")
@SpringBootTest
@RunWith(SpringRunner.class)
@AutoConfigureMessageVerifier
@EnableBinding(KafkaStreamerApplication.class)
public abstract class KafkaStreamerApplicationTests {

  @Autowired
  private DummyController dummyController;

  public void supply() {
     dummyController.handlePost();
  }
}

Контракт

org.springframework.cloud.contract.spec.Contract.make {
  label 'some_label'
  input {
   triggeredBy("supply()")
  }
  outputMessage {
    sentTo('supplier-out-0')
    body(
            anyNonBlankString()
            )
    headers {
        messagingContentType(applicationJson())
    }
  }
 }

Я пытался создать контракты для функции и потребителя, и оба работали. Я также пытался использовать поставщика типа String вместо Flux<Message<String>> и работал тоже. Вы можете посмотреть на закомментированные строки в приведенном выше коде.

Во время отладки я видел, что в StreamMessageCollectorMessageReceiver я получаю экземпляр DirectWithAttributesChannel при получении компонента MessageChannel в методе receive("supplier-out-0", 5, "SECONDS") , Также, просматривая документацию, я обнаружил реализацию MessageChannel, которая, на мой взгляд, выглядит как кандидат, которому нужно было бы создать экземпляр FluxMessageChannel.

Я предполагаю, что весенний облачный контракт ожидает отправки сообщения в очередь напрямую (DirectWithAttributesChannel), но поскольку поток уже находится в очереди, а я просто обновляю содержимое потока, он означает, что сообщение не помещается в очередь (FluxMessageChannel?), поэтому сообщение не принимается (null), и поэтому при выполнении моих тестов я получаю ошибку IllegalArgumentException с подробностями Message must not be null.

Чего-то не хватает в моей конфигурации? Может ли этот сценарий быть проверен контрактами? Это плохой подход для проверки этого с использованием контрактов?