Hợp nhất nhiều chủ đề Kafka Stream giống hệt nhau

Question

Tôi có 2 chủ đề Kafka đang phát nội dung chính xác từ nhiều nguồn khác nhau để tôi có thể có tính sẵn sàng cao trong trường hợp một trong các nguồn không thành công. Tôi đang cố gắng hợp nhất 2 chủ đề thành 1 chủ đề đầu ra bằng cách sử dụng Luồng Kafka 0.10.1.0 sao cho tôi không bỏ lỡ bất kỳ thông báo nào về lỗi và không có bản sao nào khi tất cả các nguồn đều lên.

Khi sử dụng phương pháp leftJoin của KStream, một trong các chủ đề có thể không hoạt động (chủ đề phụ), nhưng khi chủ đề chính bị hỏng, không có gì được gửi đến chủ đề đầu ra. Đây dường như là bởi vì, theo Kafka Streams developer guide,

KStream-KStream leftJoin luôn được thúc đẩy bởi các hồ sơ đến từ luồng chính

vì vậy nếu không có hồ sơ đến từ các luồng chính, nó sẽ không sử dụng các bản ghi từ luồng thứ cấp ngay cả khi chúng tồn tại. Khi luồng chính trở lại trực tuyến, đầu ra sẽ tiếp tục bình thường.

Tôi cũng đã cố gắng sử dụng outerJoin (có thêm bản ghi trùng lặp) tiếp theo là một chuyển đổi sang một KTable và groupByKey để thoát khỏi bản sao,

KStream mergedStream = stream1.outerJoin(stream2, 
    (streamVal1, streamVal2) -> (streamVal1 == null) ? streamVal2 : streamVal1, 
    JoinWindows.of(2000L)) 

mergedStream.groupByKey() 
      .reduce((value1, value2) -> value1, TimeWindows.of(2000L), stateStore)) 
      .toStream((key,value) -> value) 
      .to(outputStream) 

nhưng tôi vẫn nhận được bản sao một lần trong một thời gian. Tôi cũng đang sử dụng commit.interval.ms=200 để nhận được KTable gửi tới luồng đầu ra thường xuyên.

Cách tốt nhất để tiếp cận hợp nhất này để có được kết quả chính xác một lần từ nhiều chủ đề đầu vào giống hệt nhau là gì?

Answer 1

Sử dụng bất kỳ loại tham gia nào sẽ không giải quyết được sự cố của bạn vì bạn sẽ luôn kết thúc hoặc thiếu kết quả (tham gia bên trong trường hợp một số quầy hàng) hoặc "trùng lặp" với null (tham gia bên trái hoặc tham gia bên ngoài trường hợp cả hai luồng đều trực tuyến). Xem https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Streams+Join+Semantics để biết chi tiết về cách tham gia ngữ nghĩa trong Luồng Kafka.

Vì vậy, tôi khuyên bạn nên sử dụng API bộ xử lý mà bạn có thể kết hợp và kết hợp với DSL bằng cách sử dụng KStreamprocess(), transform() hoặc transformValues(). Xem How to filter keys and value with a Processor using Kafka Stream DSL để biết thêm chi tiết.

Bạn cũng có thể thêm cửa hàng tùy chỉnh vào bộ xử lý của mình (How to add a custom StateStore to the Kafka Streams DSL processor?) để tạo khả năng chịu lỗi trùng lặp lọc.