Spring Cloud Stream Kafka Stream Binder, MessageChannel没有创建.

Question

在我的简单测试应用程序中有一个问题。我想用kafka流绑定器创建消费者，就像这样。

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

但是当我试着测试它

@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]

当我将消费方法改为

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

没关系。

我尝试打印创建的通道

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

I receive in first case

CHANNEL: nullChannel
CHANNEL: errorChannel

第二个大小写

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

我不明白为什么会这样。Сan谁来帮帮我!

Answer 1

Kafka Streams绑定器不是MessageChannelBinder，所以它内部不使用消息通道。

您不能使用测试消息通道绑定器进行测试。