加入使用 Avro 模式定义的两个 Kafka 流时如何编写 ValueJoiner？

Question

我正在构建一个电子商务应用程序，目前我正在处理两个数据馈送：订单执行和销售中断。由于各种原因，失败的销售将是无效的执行。失败的销售将具有与订单相同的订单参考编号，因此连接是在订单参考 # 和订单项 # 上。

目前，我有两个主题 - orders 和 broken。两者都使用 Avro Schemas 定义，并使用 SpecificRecord 构建。密钥是OrderReferenceNumber。

orders 的字段：OrderReferenceNumber, Timestamp, OrderLine, ItemNumber, Quantity

broken 的字段：OrderReferenceNumber, OrderLine, Timestamp

对应的Java类是通过运行生成的

mvn clean package

我需要将orders 与broken 左连接，并在输出中包含以下字段：OrderReferenceNumber, Timestamp, BrokenSaleTimestamp, OrderLine, ItemNumber, Quantity

这是我的代码：

public static void main(String[] args) {
    // Declare variables
    final Map<String, String> avroSerdeConfig = Collections.singletonMap(KafkaAvroSerializerConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081");

    // Add Kafka Streams Properties
    Properties streamsProperties = new Properties();
    streamsProperties.put(StreamsConfig.APPLICATION_ID_CONFIG, "orderProcessor");
    streamsProperties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    streamsProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
    streamsProperties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
    streamsProperties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, SpecificAvroSerde.class);
    streamsProperties.put(KafkaAvroSerializerConfig.SCHEMA_REGISTRY_URL_CONFIG, "localhost:8081");

    // Specify Kafka Topic Names
    String orderTopic = "com.ecomapp.input.OrderExecuted";
    String brokenTopic = "com.ecomapp.input.BrokenSale";

    // Specify Serializer-Deserializer or Serdes for each Message Type
    Serdes.StringSerde stringSerde = new Serdes.StringSerde();
    Serdes.LongSerde longSerde = new Serdes.LongSerde();
    // For the Order Executed Message
    SpecificAvroSerde<OrderExecuted> ordersSpecificAvroSerde = new SpecificAvroSerde<OrderExecuted>();
    ordersSpecificAvroSerde.configure(avroSerdeConfig, false);
    // For the Broken Sale Message
    SpecificAvroSerde<BrokenSale> brokenSpecificAvroSerde = new SpecificAvroSerde<BrokenSale>();
    brokenSpecificAvroSerde.configure(avroSerdeConfig, false);


    StreamsBuilder streamBuilder = new StreamsBuilder();

    KStream<String, OrderExecuted> orders = streamBuilder
            .stream(orderTopic, Consumed.with(stringSerde, ordersSpecificAvroSerde))
            .selectKey((key, orderExec) -> orderExec.getMatchNumber().toString());
    KStream<String, BrokenSale> broken = streamBuilder
            .stream(brokenTopic, Consumed.with(stringSerde, brokenSpecificAvroSerde))
            .selectKey((key, brokenS) -> brokenS.getMatchNumber().toString());

    KStream<String, JoinOrdersExecutedNonBroken> joinOrdersNonBroken = orders
        .leftJoin(broken,
                (orderExec, brokenS) -> JoinOrdersExecutedNonBroken.newBuilder()
                        .setOrderReferenceNumber((Long) orderExec.get("OrderReferenceNumber"))
                        .setTimestamp((Long) orderExec.get("Timestamp"))
                        .setBrokenSaleTimestamp((Long) brokenS.get("Timestamp"))
                        .setOrderLine((Long) orderExec.get("OrderLine"))
                        .setItemNumber((String) orderExec.get("ItemNumber"))
                        .setQuantity((Long) orderExec.get("Quantity"))
                        .build(),
                JoinWindows.of(TimeUnit.MILLISECONDS.toMillis(1))
                Joined.with(stringSerde, ordersSpecificAvroSerde, brokenSpecificAvroSerde))
        .peek((key, value) -> System.out.println("key = " + key + ", value = " + value));


    KafkaStreams orderStreams = new KafkaStreams(streamBuilder.build(), streamsProperties);
    orderStreams.start();

    // print the topology
    System.out.println(orderStreams.localThreadsMetadata());

    // shutdown hook to correctly close the streams application
    Runtime.getRuntime().addShutdownHook(new Thread(orderStreams::close));

}

当我运行它时，我得到以下 maven 编译错误：

[ERROR] /Tech/Projects/jCom/src/main/java/com/ecomapp/kafka/orderProcessor.java:[96,26] incompatible types: cannot infer type-variable(s) VO,VR,K,V,VO
    (argument mismatch; org.apache.kafka.streams.kstream.Joined<K,V,com.ecomapp.input.BrokenSale> cannot be converted to org.apache.kafka.streams.kstream.Joined<java.lang.String,com.ecomapp.OrderExecuted,com.ecomapp.input.BrokenSale>)

真正的问题在于定义我的ValueJoiner。当涉及 Avro 模式时，Confluent 文档不太清楚如何执行此操作（我也找不到示例）。什么是正确的定义方式？

Answer 1

不确定为什么 Java 无法解析该类型。

试试：

Joined.<String,OrderExecuted,BrokenSale>with(stringSerde, ordersSpecificAvroSerde, brokenSpecificAvroSerde))

明确指定类型。