Spring Cloud Stream Kafka Stream Binder, MessageChannel is not created

Question

I have a problem in my simple test application. I'd like create consumer with kafka stream binder like this.

@SpringBootApplication
public class CloudStreamAggregatorApplication {

  public static void main(String[] args) {
    SpringApplication.run(CloudStreamAggregatorApplication.class, args);
  }
  @Bean
  public Consumer<KStream<String,String>> consume() {
    return input -> input.foreach((k,v) -> System.out.println("CONSUMER: "+v));
  }
}

But when i try to test it

@SpringBootTest
@EmbeddedKafka
@Import(TestChannelBinderConfiguration.class)
@DirtiesContext
class CloudStreamConsumerApplicationTests {

  @Autowired
  private InputDestination input;

  @Test
  void test01_Consume() {
    input.send(new GenericMessage<>("test"));
  }
}

I receive exception

java.lang.IndexOutOfBoundsException: Index 0 out of bounds for length 0
  at java.base/jdk.internal.util.Preconditions.outOfBounds(Preconditions.java:64) ~[na:na]
  at java.base/jdk.internal.util.Preconditions.outOfBoundsCheckIndex(Preconditions.java:70) ~[na:na]
  at java.base/jdk.internal.util.Preconditions.checkIndex(Preconditions.java:248) ~[na:na]
  at java.base/java.util.Objects.checkIndex(Objects.java:359) ~[na:na]
  at java.base/java.util.ArrayList.get(ArrayList.java:427) ~[na:na]
  at org.springframework.cloud.stream.binder.test.AbstractDestination.getChannel(AbstractDestination.java:34) ~[spring-cloud-stream-3.1.0-test-binder.jar:3.1.0]

And when i change consume method to

@Bean
public Consumer<Flux<String>> consume() {
    return f -> f.subscribe(p -> System.out.println("CONSUMER: "+p));
}

It's ok.

I try to print created channels

@Autowired
private Map<String,MessageChannel> channels;
...
channels.forEach((k,v) -> System.out.println("CHANNEL: "+k));

I receive in first case

CHANNEL: nullChannel
CHANNEL: errorChannel

And in second case

CHANNEL: nullChannel
CHANNEL: errorChannel
CHANNEL: consume-in-0
CHANNEL: test.anonymous.errors

I didn't get why it happens. Сan anyone help me!

Answer 1

The Kafka Streams binder is not a MessageChannelBinder so it doesn't use message channels internally.

You can't test with the test messsage channel binder.