В моем Kafka Topi c конец смещения выше, чем номер смещения последних сообщений, из-за этого я могу видеть отставание потребителя 1

Question

Использование приложения Kafka Stream Я читаю сообщения из input-topi c и публикую в Output-topi c.

Но в моем Output-topi c последнее сообщение Offset значение равно "18", но значение конца смещения равно "20"

Topic test.topic.out[33], offset: 17, key: 228, payload: 1003 bytes: {"ITEM_LOC_CHG_ID":4657706,"sku_selling_location_id":"17960858808","selling_location_id":"7619017","event_type":"P","store_number":"228","sku":"7734279","location_indicator":"1","capacity":"0.0","width":"0.0","last_update_userid":"sslpromospaceupdate1","sku_selling_location_change_date":"2019-12-17 13:08:49.0","ticket_processed":"N","store_item_shelf_sequence":null,"store_section_short_name":null,"mezzanine":null,"store_aisle_id_nearest":null,"store_segment_number_actual":null,"store_aisle_id":null,"store_reference_bay_id":null,"store_logical_bay_id":null,"store_shelf_id":null,"store_plinth_id":null,"online_processed":"N","item_loc_chg_txn_id":"402141004","sub_category":"330","default_pack_size":"4.0","selling_location_type":"0","product_sequence_within_subcat":"11","old_capacity":null,"old_width":null,"old_store_aisle_id":null,"old_store_logical_bay_id":null,"old_store_shelf_id":null,"old_store_item_shelf_sequence":null,"old_product_sequence_within_subcat":null,"COPY_TIME":1576588219061}

Topic test.topic.out[33], offset: 18, key: 228, payload: 1003 bytes: {"ITEM_LOC_CHG_ID":4657707,"sku_selling_location_id":"17960858809","selling_location_id":"7619017","event_type":"P","store_number":"228","sku":"7860138","location_indicator":"1","capacity":"0.0","width":"0.0","last_update_userid":"sslpromospaceupdate1","sku_selling_location_change_date":"2019-12-17 13:08:49.0","ticket_processed":"N","store_item_shelf_sequence":null,"store_section_short_name":null,"mezzanine":null,"store_aisle_id_nearest":null,"store_segment_number_actual":null,"store_aisle_id":null,"store_reference_bay_id":null,"store_logical_bay_id":null,"store_shelf_id":null,"store_plinth_id":null,"online_processed":"N","item_loc_chg_txn_id":"402141005","sub_category":"330","default_pack_size":"6.0","selling_location_type":"0","product_sequence_within_subcat":"12","old_capacity":null,"old_width":null,"old_store_aisle_id":null,"old_store_logical_bay_id":null,"old_store_shelf_id":null,"old_store_item_shelf_sequence":null,"old_product_sequence_within_subcat":null,"COPY_TIME":1576588219061}
% Reached end of topic test.topic.out [33] at offset 20: exiting

, из-за этого мое потребительское отставание всегда показывает 1 (даже если мой потребительский сервис использовал все сообщения из раздела ).

в чем может быть root причина проблемы?

Answer 1

Я думаю, что вы используете транзакционную семантику в вашем производителе. Когда ваш производитель использует транзакции, kafka генерирует « контрольных пакетов »

Контрольный пакет содержит одну запись, называемую контрольной записью. Контрольные записи не должны передаваться приложениям. Вместо этого они используются потребителями для фильтрации прерванных транзакционных сообщений.

Проверьте официальную команду c: http://kafka.apache.org/documentation/#controlbatch