Phương thức thăm dò ý kiến ​​của người tiêu dùng Kafka() bị chặn

Question

Tôi mới sử dụng Kafka 0.9 và thử nghiệm một số tính năng tôi nhận thấy một hành vi lạ trong Java thực hiện Người tiêu dùng (KafkaConsumer).

Nhà môi giới Kafka nằm trong máy bên ngoài Ambari.

Ngay cả khi tôi có thể triển khai Nhà sản xuất và bắt đầu gửi tin nhắn đến nhà môi giới bên ngoài, tôi không biết tại sao, khi người tiêu dùng cố đọc các sự kiện (bình chọn), nó bị kẹt.

Tôi biết nhà sản xuất đang hoạt động tốt, vì tôi có thể tiêu thụ tin nhắn thông qua bảng điều khiển người tiêu dùng (đang hoạt động cục bộ trên ambari). Nhưng khi tôi thực thi Java Consumer, không có gì xảy ra, chỉ bị mắc kẹt. Gỡ lỗi mã Tôi có thể thấy rằng nó bị chặn tại đường dây poll():

ConsumerRecords<String, String> records = consumer.poll(100); 

Thời gian chờ không có gì, bằng cách này. Không quan trọng nếu bạn đặt 0, 100 hoặc 1000 ms, người tiêu dùng bị chặn trong dòng này và không hết thời gian chờ và cũng không loại trừ ngoại lệ.

tôi đã cố gắng tất cả các loại tài sản thay thế, chẳng hạn như advertised.host.name, advertised.listener ... và như vậy, có zero may mắn.

Mọi trợ giúp sẽ được đánh giá cao. Cảm ơn trước!

Answer 1

Lý do có thể là máy mà mã người tiêu dùng của bạn đang chạy không thể kết nối với người quản lý vườn thú. Hãy thử chạy cùng một mã người tiêu dùng trên máy nơi Kafka của bạn được cài đặt (tôi đã thử điều này và làm việc cho tôi). Tôi cũng giải quyết vấn đề bằng cách đề cập đến các thuộc tính dưới đây trong file server.properties: // trong trường hợp của tôi là ip công cộng của máy ec2, tôi có kafka và zookeeper được cài đặt trên cùng ec2. advertised.port=9092 ConsumerRecords<String, String> records = consumer.poll(100); Tuyên bố trên không có nghĩa là người tiêu dùng sẽ hết thời gian sau 100 mili giây, đó là khoảng thời gian bỏ phiếu. Bất cứ dữ liệu nào nó chụp trong 100 ms được đọc vào bộ sưu tập bản ghi.