如何捕获与Kafka流连接条件不匹配的Kafka记录？

Question

我正在通过连接kstream和KTable来丰富数据。kstream包含车辆发送的消息，ktable包含车辆数据。我遇到的问题是，我想从表中没有相应联接键的流捕获消息。Kafka流静默跳过它们没有连接匹配的记录。是否有某种方法可以将这些记录发送到不同的主题，以便以后可以对它们进行处理？

StreamsBuilder builder = new StreamsBuilder();
        final KTable<String, VinMappingInfo> vinMappingTable = builder.table(vinInfoTopic, Consumed.with(Serdes.String(), valueSerde));
        KStream<String, VehicleMessage> vehicleStream = builder.stream(sourceTopic);
        vehicleStream.join(vinMappingTable, (vehicleMsg, vinInfo) -> {
            log.info("joining {} with vin info {}", vehicleMsg.getPayload().getId(), vinInfo.data.vin);
            vehicleMsg.setVin(vinInfo.data.vin);
            return vehicleMsg;
        }, Joined.with(null, null, valueSerde))
                .to(destinationTopic);

        final Topology topology = builder.build();
        log.info("The topology of connected processor nodes: \n {}", topology.describe());
        KafkaStreams streams = new KafkaStreams(topology, config);
        streams.cleanUp();
        streams.start();

Answer 1

可以使用左联接：

java prettyprint-override">stream.leftJoin(table,...);

这确保输入流中的所有记录都在输出流中。对于这种情况，将使用apply(streamValue,null)调用ValueJoiner。