Tác giả: Danica Fine & Nikoleta Verbeck | Ngày đăng: 15/11/2022

Link bài viết gốc | Link series tiếng Việt

Khi gặp sự cố với Apache Kafka® - ví dụ như số lượng kết nối tới các broker tăng đột biến hoặc việc gom nhóm bản ghi (record batching) hoạt động không bình thường - các bạn sẽ rất dễ coi các vấn đề này như những vấn đề cần được giải quyết riêng lẻ. Tuy nhiên, bạn sẽ sớm nhận ra, phần lớn các vấn đề này thực chất chỉ là triệu chứng của một vấn đề lớn hơn. Thay vì chỉ xử lý từng triệu chứng riêng lẻ, chẳng phải sẽ tốt hơn nếu ta đi đến tận gốc vấn đề bằng một chẩn đoán chuẩn chỉ?

Nếu bạn đang muốn nâng cao kỹ năng debug Kafka của mình và hiểu sâu hơn về các vấn đề phổ biến cũng như các “bệnh lý” tiềm ẩn đằng sau từng triệu chứng, thì loạt bài viết này là dành cho bạn.

Các triệu chứng có trong loạt bài viết

Trong loạt bài viết này, chúng ta sẽ cùng tìm hiểu một số triệu chứng phổ biến bạn có thể gặp phải khi sử dụng Kafka, bao gồm:

• Giảm thông lượng gửi message
• Tăng request rate, thời gian response và/hoặc tăng tải trên broker
• Tăng thời gian tái cân bằng (rebalance) consumer (bài viết này)
• Tăng số lượng kết nối

Những vấn đề này khá phổ biến, đến mức đôi khi bạn sẽ không để ý tới chúng nếu chúng chưa gây ảnh hưởng nghiêm trọng tới hoạt động thông thường. Hãy cùng khám phá từng triệu chứng cụ thể, tìm hiểu xem chúng là gì và tác động của chúng, cũng như đặt ra những câu hỏi giúp bạn xác định gốc rễ vấn đề.

Nội dung bài viết này: Tăng thời gian tái cân bằng (rebalance) consumer (Increased consumer rebalance time)

Tái cân bằng (rebalancing) diễn ra trong Kafka khi có consumer tham gia hoặc rời khỏi một consumer group. Khi đó, số lượng consumer thay đổi → các topic-partition phải được phân phối lại cho phù hợp.

Các loại tái cân bằng

Thoạt nhìn, rebalancing trông có vẻ đơn giản: thay đổi số lượng consumer → phân phối lại topic-partition. Nhưng thực tế thì phức tạp hơn nhiều và tùy thuộc vào loại tái cân bằng đang sử dụng.

1. Stop-the-World

Đây là loại rebalance phổ biến nhất.

0. Giả sử bạn có một consumer group đã ổn định, với các topic-partition đã được phân bổ cho từng consumer. Mỗi consumer có member.id, và một trong số đó là group leader (thường là consumer tham gia đầu tiên). Khi một consumer mới muốn tham gia, nó gửi JoinGroup request đến Group Coordinator, kèm theo danh sách topic nó muốn đăng ký (subscribe).

1. Group Coordinator nhận request mới → bắt đầu quy trình rebalance. Coordinator yêu cầu tất cả các consumer hiện tại để gửi lại JoinGroup request. Thông báo này được gắn trong phản hồi của heartbeat mà consumer gửi định kỳ.

2. Mỗi consumer trong group có khoảng thời gian max.poll.interval.ms để hoàn tất xử lý đang dở dang hiện tại và gửi JoinGroup request của mình, tại thời điểm đó, "Thế giới bị dừng lại" (stop-the-world). Với tất cả các JoinGroup request, Group Coordinator xác định được tất cả các consumer trong group và các topic nào thuộc về consumer group. Nó gửi JoinResponses đến các thành viên, chọn một group leader từ các consumer thành viên, và để group leader tính toán việc phân bổ phân vùng (partition assignment).

3. Tất cả các consumer thành viên trong group phản hồi bằng SyncGroup request. Group leader gửi danh sách các phân bổ phân vùng (partition assignments) của mình cùng với request này.

4. Group Coordinator gửi SyncResponse đến từng consumer, xác nhận các topic-partition được phân bổ cho chúng.

5. Cuối cùng, các consumer xác nhận phân bổ của chúng và quá trình xử lý tiếp tục. "Thế giới hoạt động trở lại".

Nếu chỉ có một consumer được thêm vào hoặc loại bỏ khỏi consumer group, quá trình này khá dễ theo dõi. Tuy nhiên, khi cả ứng dụng, ví dụ như các ứng dụng với nhiều instance Kafka Streams, được triển khai lại toàn bộ như một phần của quá trình nâng cấp, tất cả các consumer có thể được nâng cấp và khởi động lại cùng một lúc hoặc trong một khoảng thời gian ngắn. Điều này có thể dẫn đến hiệu ứng domino, nhiều JoinGroup request khiến rebalance xảy ra nhiều lần, do đó làm cho thời gian rebalance kéo dài lâu hơn so với bình thường.

Thời gian rebalance càng lâu, độ trễ của consumer (consumer lag) càng lớn khi tất cả các consumer hoạt động trở lại. Để giảm thiểu độ trễ này, điều quan trọng là quá trình rebalance phải diễn ra trong thời gian ngắn nhất có thể để không làm gián đoạn luồng tiêu thụ và xử lý dữ liệu hiện tại của các consumer hiện có. May mắn thay, có những tùy chọn khác có sẵn để giúp giảm các tác động của rebalance.

2. Incremental Cooperative

Trong cơ chế Incremental Cooperative, các consumer ít bị ảnh hưởng hơn bởi sự thay đổi trong consumer group, và có thể tiếp tục xử lý dữ liệu trong khi rebalance diễn ra.

Các bước đầu (0–2) giống với cơ chế Stop-the-World.

Note từ dịch giả:
Tóm tắt bước 0-2:

0. Consumer mới gửi JoinGroup request cho Coordinator.
1. Group Coordinator yêu cầu các consumer hiện tại cũng gửi JoinGroup request.
2. Mỗi consumer có thời gian max.poll.interval.ms để hoàn tất xử lý hiện tại và gửi request. Coordinator xác định toàn bộ các consumer trong group và các topic liên quan. Gửi JoinResponse và chọn 1 leader để tính toán lại việc phân bổ partition.

3. Mỗi consumer trong group có khoảng thời gian max.poll.interval.ms để hoàn tất xử lý hiện tại và gửi JoinGroup request. Lần này, họ cũng gửi kèm phân bổ topic-partition hiện tại của mình. Và thay vì "dừng toàn bộ thế giới" (stop-the-world) để chờ phân bổ mới, các consumer được tự do tiếp tục xử lý các topic-partition đã được gán cho đến khi có thông báo tiếp theo.

4. Group Coordinator lại sử dụng các JoinGroup request để xác định tất cả các consumer và các topic nào nên thuộc về consumer group. Nó gửi JoinResponses đến các thành viên, chọn một group leader, và để leader tính toán việc phân bổ partition.

5. Tất cả các consumer thành viên trong group phản hồi bằng SyncGroup request. Group leader gửi các phân bổ partition đã cập nhật cùng với request của mình.

6. Group Coordinator xem xét các phân bổ partition mới, so sánh chúng với phân bổ cũ, và chỉ gửi SyncResponses đến các consumer có phân bổ partition thay đổi. Do đó, chỉ các consumer bị ảnh hưởng mới phải dừng lại để tái cân bằng.

7. Cuối cùng, các consumer được cập nhật sẽ xác nhận phân bổ mới của họ, và toàn bộ group tiếp tục quá trình xử lý.

3. Static Group Assignment

Một số trường hợp có thể dùng Static Assignment bằng cách gán group.instance.id cho từng consumer.

• Nếu một consumer tạm thời ngắt kết nối (trong giới hạn session.timeout.ms) rồi quay lại, KHÔNG tái cân bằng xảy ra.
• Điều này giảm thiểu tối đa gián đoạn, đặc biệt với các nhóm consumer cố định.

Trong một số trường hợp, người dùng có thể giảm thiểu tối đa tác động của rebalance bằng cách sử dụng Static Group Assignment. Mỗi consumer trong consumer group được gán một group.instance.id và một phân bổ partition tĩnh (không thay đổi). Khi một consumer ngừng hoạt động vì bất kỳ lý do gì, sẽ không xảy ra rebalance miễn là consumer với group.instance.id đó quay lại hoạt động trong khoảng thời gian session.timeout.ms (có thể cấu hình).

Note từ dịch giả:
Phân bổ partition tĩnh là một tập hợp partition cố định, phân bổ này không thay đổi trừ khi cấu hình topic thay đổi (ví dụ thêm partition)

Các chỉ số (metrics) cần theo dõi

Việc hiểu rõ tình trạng sức khỏe của các consumer là bước đầu tiên để giảm thời gian tái cân bằng (rebalance) kéo dài và tìm ra nguyên nhân gốc rễ gây ra vấn đề. Không có gì ngạc nhiên khi các chỉ số theo dõi consumer được thiết kế để hỗ trợ chính xác cho việc này.

• join-rate
kafka.consumer:type=consumer-coordinator-metrics,name=join-rate
  Số lượng JoinGroup request được gửi mỗi giây trong một consumer group. Đây là số liệu về consumer group. Khi rebalance xảy ra, tham gia (joining) là bước đầu tiên của quá trình, vì vậy đây là một số liệu đáng để theo dõi. Đối với một consumer group khỏe mạnh, tỷ lệ tham gia (join-rate) nên ở mức thấp; một số lượng lớn request mỗi giây có thể cho thấy consumer group không ổn định.

• records-lag-max
  Độ trễ xử lý record tối đa trên tất cả các partition của một consumer group hoặc từng partition riêng lẻ.

  • Cấp độ consumer group: kafka.consumer:type=consumer-fetch-manager-metrics,name=records-lag-max
  • Cấp độ partition: kafka.consumer:type=consumer-fetch-manager-metrics,partition="{partition}",topic="{topic}"

  Độ trễ xử lý record tối đa ở mức cao hoặc đang tăng cho thấy consumer group không thể theo kịp các record được producer gửi đến hoặc một partition cụ thể đang bị ảnh hưởng. Bản thân số liệu này không chỉ ra chính xác nguyên nhân gây ra độ trễ, nhưng nó là một số liệu quan trọng cần lưu ý, đặc biệt nếu bạn nghi ngờ rằng các quá trình rebalance của mình đang mất quá nhiều thời gian.

Tiếp tục chẩn đoán

Ngoài việc nhận thấy thời gian tái cân bằng (rebalance) tăng lên...

• ... bạn có thấy mức sử dụng bộ nhớ cao hơn bình thường, kích thước consumer group lớn, và số lượng kết nối tăng không?
  Có thể bạn đang tạo mỗi thread một consumer trong cùng một service instance (). Mục đích ban đầu là cố gắng tăng và song song hóa việc tiêu thụ dữ liệu trong một instance duy nhất. Đương nhiên, để làm điều này, bạn sẽ phải triển khai đa luồng (multithread) và khởi tạo một instance consumer trong mỗi luồng (thread) đó. Tuy nhiên, cái giá phải trả cho việc tăng số lượng consumer chính là những gì chúng ta đã thấy ở trên.

  Để chuyển sang mô hình đa luồng, bạn nên dùng một consumer duy nhất, tách nó ra một luồng riêng đọc dữ liệu rồi chuyển sang nhóm luồng (thread pool) riêng để xử lý song song. Có nhiều cách để thực hiện điều này thành công, nhưng hãy nhớ rằng consumer không thread-safe, nên cần triển khai cẩn thận.

• ... bạn có thấy consumer group lớn và số lượng kết nối tăng không?
  Triệu chứng này tương tự như cái được mô tả trong phần cuối của vấn đề giảm hiệu suất batching. Nếu Kafka cluster của bạn chạy trên cloud, các instance KafkaConsumer của bạn có thể được định cỡ quá nhỏ. Nên xem xét lại cấu hình workload của KafkaConsumer để scale hợp lý (vertical scaling).

• … bạn có rơi vào ngõ cụt không?
  Phải thừa nhận rằng hầu hết các số liệu và giải thích về rebalance ở trên đều liên quan đến client và cách chúng hoạt động. Điều đó không có nghĩa là không thể điều chỉnh từ phía server, cụ thể là từ Group Coordinator. Bạn có thể thấy các rebalance mất nhiều thời gian hơn nếu, chẳng hạn, Group Coordinator phải quản lý quá nhiều group hoặc nếu một partition quá lớn (nghĩa là mất nhiều thời gian hơn để tải (load)). Nếu bạn gặp ngõ cụt khi theo dõi các vấn đề rebalance từ phía client, hãy kiểm tra Group Coordinator của bạn.

Kết luận

Bùng nổ rebalance có thể khiến bạn hoang mang nếu bạn không biết phải theo dõi từ đâu khi nó xảy ra. May mắn thay, việc hiểu rõ các loại rebalancing, chỉ số quan trọng và biết đặt những câu hỏi đúng sẽ khiến quá trình này bớt đáng sợ hơn vào lần tới.