Spring的云流Kafka流活页夹3。x：在多个输出绑定的情况下，没有输出到第二个输出主题

Question

我有以下处理器bean方法签名：

@Bean
public BiFunction<KStream<String, MyClass>, KStream<String, String>, KStream<String, MyClass>[]> myStream() {
    return (inputStream1, intputStream2) -> {

        intputStream2
            .peek((k, v) -> {
                log.debug(...);
            });

        return inputStream1
            .mapValues(...)
            .branch((k,v) -> true, (k,v) -> true);

    };
}

相关物业：

spring.cloud.stream.function.definition: ...;myStream
spring.cloud.stream.bindings:
  myStream-in-0:
    destination: inp0
  myStream-in-1:
    destination: inp1
  myStream-out-0:
    destination: out0
  myStream-out-1:
    destination: out1

Spring Cloud Kafka Stream版本Hoxton. SR4（Spring-Cloud-stream-binder-kafka-stream： jar： 3.0.4. RELEASE），嵌入式Kafka版本2.5.0。

我正在使用嵌入式Kafka测试我的拓扑：

@RunWith(SpringRunner.class)
@SpringBootTest(
    properties = "spring.cloud.stream.kafka.binder.brokers=${spring.embedded.kafka.brokers}"
)
@EmbeddedKafka(partitions = 1,
        topics = {
            "inp0", "inp1", "out0", "out1"
        },
        brokerPropertiesLocation = "kafka.properties"
)
@Slf4j
public class MyApplicationTests {

    @Test
    public void embeddedKafkaTest() throws IOException, InterruptedException {
        Consumer<String, MyClass> out0Consumer = createConsumer("out0ConsumerGroup");
        Consumer<String, MyClass> out1Consumer = createConsumer("out1ConsumerGroup");

        this.embeddedKafka.consumeFromAnEmbeddedTopic(out0Consumer, "out0");
        this.embeddedKafka.consumeFromAnEmbeddedTopic(out1Consumer, "out1");

        latch = new CountDownLatch(1);
        // ... publish ...
        latch.await(15, TimeUnit.SECONDS);

        ConsumerRecords<String, MyClass> out0 = KafkaTestUtils.getRecords(out0Consumer);
        assertThat(out0.count(), is(greaterThanOrEqualTo(1)));

        ConsumerRecords<String, MyClass> out1 = KafkaTestUtils.getRecords(out1Consumer);
        assertThat(out1.count(), is(greaterThanOrEqualTo(1)));

    }

private <K,V> Consumer<K, V> createConsumer(String groupName) {
    Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(groupName, "true", this.embeddedKafka);
    consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
    return new DefaultKafkaConsumerFactory<K, V>(consumerProps).createConsumer();
}

我的测试显示，来自myStream的消息按预期到达并到达主题“out0”中，但“out1”主题仍然为空，单元测试在第二次断言中失败。

我已经尝试了一些东西，但是看起来第二个输出主题的输出根本没有产生（第一个输出主题的输出产生得很好）。

你能看到我的设置有什么错误吗？

还有一件事：myStream bean方法定义中的return语句显示了一个编译器警告：

为varargs参数创建未检查的泛型数组

但是看起来这就是Spring Cloud Kafka Stream 3. x API要求定义返回类型的方式？

Answer 1

您将两个谓词传递给分支方法，它们的计算结果总是true。第一个谓词始终获胜，并向第一个输出绑定生成数据。分支方法调用在第一个谓词求值为true后退出。有关更多详细信息，请参阅javadoc。您应该使用不同的谓词（可能会检查键/值的某些条件）。如果第一个谓词失败，第二个谓词成功，那么您将看到生成到第二个输出主题的数据。

关于编译器警告，我认为您可以安全地忽略它，因为API将确保传递到分支调用的谓词对象具有正确的类型。由于该方法的实现使用泛型varargs，因此会出现该异常。有关编译器警告的详细信息，请参阅此线程。