search

flier / tokio-kafka

Asynchronous Rust client for Apache Kafka
Apache License 2.0
31 stars 1 forks source link

Prefetch record base on low/high watermark #31

Open flier opened 6 years ago

flier commented 6 years ago

solve #29

polachok commented 6 years ago

With new group it's stuck on "fetching records of partitions"

DEBUG 2018-05-10T08:41:11Z: tokio_kafka::client::service: received message: Message::WithoutBody(OffsetFetch(OffsetFetchResponse { header: ResponseHeader { correlation_id: 5 }, topics: [OffsetFetchTopicStatus
 { topic_name: "<redacted>", partitions: [OffsetFetchPartitionStatus { partition_id: 0, offset: -1, metadata: Some(""), error_code: 0 }, OffsetFetchPartitionStatus { partition_id: 1, offset: -1, metadata
: Some(""), error_code: 0 }, OffsetFetchPartitionStatus { partition_id: 2, offset: -1, metadata: Some(""), error_code: 0 }, OffsetFetchPartitionStatus { partition_id: 3, offset: -1, metadata: Some(""), error_
code: 0 }, OffsetFetchPartitionStatus { partition_id: 4, offset: -1, metadata: Some(""), error_code: 0 }, OffsetFetchPartitionStatus { partition_id: 5, offset: -1, metadata: Some(""), error_code: 0 }, OffsetF
etchPartitionStatus { partition_id: 6, offset: -1, metadata: Some(""), error_code: 0 }, OffsetFetchPartitionStatus { partition_id: 7, offset: -1, metadata: Some(""), error_code: 0 }, OffsetFetchPartitionStatu
s { partition_id: 8, offset: -1, metadata: Some(""), error_code: 0 }, OffsetFetchPartitionStatus { partition_id: 9, offset: -1, metadata: Some(""), error_code: 0 }, OffsetFetchPartitionStatus { partition_id: 
10, offset: -1, metadata: Some(""), error_code: 0 }, OffsetFetchPartitionStatus { partition_id: 11, offset: -1, metadata: Some(""), error_code: 0 }, OffsetFetchPartitionStatus { partition_id: 12, offset: -1, 
metadata: Some(""), error_code: 0 }, OffsetFetchPartitionStatus { partition_id: 13, offset: -1, metadata: Some(""), error_code: 0 }, OffsetFetchPartitionStatus { partition_id: 14, offset: -1, metadata: Some("
"), error_code: 0 }, OffsetFetchPartitionStatus { partition_id: 15, offset: -1, metadata: Some(""), error_code: 0 }] }] }))
TRACE 2018-05-10T08:41:11Z: tokio_kafka::consumer::subscribed: fetching records of partitions: []
TRACE 2018-05-10T08:41:11Z: tokio_kafka::consumer::subscribed: fetching records of partitions: []
TRACE 2018-05-10T08:41:11Z: tokio_kafka::consumer::subscribed: fetching records of partitions: []
TRACE 2018-05-10T08:41:11Z: tokio_kafka::consumer::subscribed: fetching records of partitions: []
TRACE 2018-05-10T08:41:11Z: tokio_kafka::consumer::subscribed: fetching records of partitions: []
TRACE 2018-05-10T08:41:11Z: tokio_kafka::consumer::subscribed: fetching records of partitions: []
TRACE 2018-05-10T08:41:11Z: tokio_kafka::consumer::subscribed: fetching records of partitions: []
TRACE 2018-05-10T08:41:11Z: tokio_kafka::consumer::subscribed: fetching records of partitions: []
TRACE 2018-05-10T08:41:11Z: tokio_kafka::consumer::subscribed: fetching records of partitions: []
TRACE 2018-05-10T08:41:11Z: tokio_kafka::consumer::subscribed: fetching records of partitions: []
TRACE 2018-05-10T08:41:11Z: tokio_kafka::consumer::subscribed: fetching records of partitions: []
TRACE 2018-05-10T08:41:11Z: tokio_kafka::consumer::subscribed: fetching records of partitions: []
TRACE 2018-05-10T08:41:11Z: tokio_kafka::consumer::subscribed: fetching records of partitions: []
TRACE 2018-05-10T08:41:11Z: tokio_kafka::consumer::subscribed: fetching records of partitions: []
TRACE 2018-05-10T08:41:11Z: tokio_kafka::consumer::subscribed: fetching records of partitions: []
TRACE 2018-05-10T08:41:11Z: tokio_kafka::consumer::subscribed: fetching records of partitions: []
TRACE 2018-05-10T08:41:11Z: tokio_kafka::consumer::subscribed: fetching records of partitions: []
TRACE 2018-05-10T08:41:11Z: tokio_kafka::consumer::subscribed: fetching records of partitions: []
TRACE 2018-05-10T08:41:11Z: tokio_kafka::consumer::subscribed: fetching records of partitions: []
TRACE 2018-05-10T08:41:11Z: tokio_kafka::consumer::subscribed: fetching records of partitions: []
TRACE 2018-05-10T08:41:11Z: tokio_kafka::consumer::subscribed: fetching records of partitions: []
TRACE 2018-05-10T08:41:11Z: tokio_kafka::consumer::subscribed: fetching records of partitions: []
TRACE 2018-05-10T08:41:11Z: tokio_kafka::consumer::subscribed: fetching records of partitions: []
TRACE 2018-05-10T08:41:11Z: tokio_kafka::consumer::subscribed: fetching records of partitions: []
TRACE 2018-05-10T08:41:11Z: tokio_kafka::consumer::subscribed: fetching records of partitions: []
TRACE 2018-05-10T08:41:11Z: tokio_kafka::consumer::subscribed: fetching records of partitions: []
TRACE 2018-05-10T08:41:11Z: tokio_kafka::consumer::subscribed: fetching records of partitions: []
TRACE 2018-05-10T08:41:11Z: tokio_kafka::consumer::subscribed: fetching records of partitions: []
flier commented 6 years ago

You may try the record_format_v2 branch which fixed a similar issue

polachok commented 6 years ago

Cherry-picked that commit, same problem