Baike.dev
Log in
Back to tool/Back to issues
#604·faust

NoAck and Manual Acknowledgement not working in case of Exception in agent and Rebalancing

Author: ankushasthCreated Jun 2, 2020Updated Aug 22, 2023

Checklist

  • I have included information about relevant versions
  • I have verified that the issue persists when using the master branch of Faust.

Steps to reproduce

I have an agent where I have used noack() to stop automatic acknowledgement. From inside that agent I have raised an exception. The exception causes the agent to restart. Before exception if the offset was (partition 0: offset 1112) then after restart it becomes (partition 0: offset 1113). It is still moving to the next offset.

In case of rebalancing it is skipping offsets and we are losing data. Before rebalancing if the offset was (partition 0: offset 1112) then after rebalancing it becomes (partition 0: offset 1115).

Below is my code:

pytb
    async for event in stream.noack().events():
        # Create event model, log event details and publish metrics.
        event_model = BaseProcessor.create_event_model(event)
        BaseProcessor.log_event_details(event_model)
        raise Exception("hey!!!!!!!")
        event.ack()
        yield None

Expected behavior

After exception/rebalancing the offset should not be skipped and agent should start from the same offset during whose processing exception/rebalancing happened as we has stopped automatic acknowledgement and exception/rebalancing prevented manual acknowledgement from happening.

Actual behavior

After exception/rebalancing the offset get skipped and agent does not start from the same offset during whose processing exception/rebalancing happened.

Before exception if the offset was (partition 0: offset 1112) then after restart it becomes (partition 0: offset 1113). It is still moving to the next offset.

In case of rebalancing it is skipping offsets and we are losing data. Before rebalancing if the offset was (partition 0: offset 1112) then after rebalancing it becomes (partition 0: offset 1115).

Full traceback

pytb
2020/06/02 00:12:47.173 INFO  [di_cust_experience_conversation_processor.processors.conversation_processor] (print_stats) Rebalancing: False, Rebalances: 1, Events: 22, Messages: 22, Sent: 0, Messages/s 0
2020/06/02 00:12:47.173 INFO  [di_cust_experience_conversation_processor.processors.conversation_processor] (print_stats)     TP(topic='com.expedia.vap.Conversation.gcolab', partition=0), read:,             , committed:             , end:             
2020/06/02 00:12:47.173 INFO  [di_cust_experience_conversation_processor.processors.conversation_processor] (print_stats)     TP(topic='com.expedia.vap.Conversation.gcolab', partition=1), read:,             , committed:             , end:             
2020/06/02 00:12:47.173 INFO  [di_cust_experience_conversation_processor.processors.conversation_processor] (print_stats)     TP(topic='com.expedia.vap.Conversation.gcolab', partition=2), read:,             , committed:             , end:             
2020/06/02 00:12:47.174 INFO  [di_cust_experience_conversation_processor.processors.conversation_processor] (print_stats)     TP(topic='com.expedia.vap.Conversation.gcolab', partition=3), read:,       428490, committed:       428490, end:      1032411
2020/06/02 00:12:47.174 INFO  [di_cust_experience_conversation_processor.processors.conversation_processor] (print_stats)     TP(topic='com.expedia.vap.Conversation.gcolab', partition=4), read:,             , committed:             , end:             
2020/06/02 00:12:47.174 INFO  [di_cust_experience_conversation_processor.processors.conversation_processor] (print_stats)     TP(topic='com.expedia.vap.Conversation.gcolab', partition=5), read:,             , committed:             , end:             
2020/06/02 00:12:47.174 INFO  [di_cust_experience_conversation_processor.processors.conversation_processor] (print_stats)     TP(topic='com.expedia.vap.domain.entities.ConversationSegmentReprocess.gcolab', partition=0), read:,             , committed:             , end:             
2020/06/02 00:12:47.175 INFO  [di_cust_experience_conversation_processor.processors.conversation_processor] (print_stats)     TP(topic='com.expedia.vap.domain.entities.ConversationSegmentReprocess.gcolab', partition=1), read:,             , committed:             , end:             
2020/06/02 00:12:47.175 INFO  [di_cust_experience_conversation_processor.processors.conversation_processor] (print_stats)     TP(topic='com.expedia.vap.domain.entities.ConversationSegmentReprocess.gcolab', partition=2), read:,             , committed:             , end:             
2020/06/02 00:12:47.175 INFO  [di_cust_experience_conversation_processor.processors.conversation_processor] (print_stats)     TP(topic='com.expedia.vap.domain.entities.ConversationSegmentReprocess.gcolab', partition=3), read:,             , committed:             , end:             
2020/06/02 00:12:47.175 INFO  [di_cust_experience_conversation_processor.processors.conversation_processor] (print_stats)     TP(topic='com.expedia.vap.domain.entities.ConversationSegmentReprocess.gcolab', partition=4), read:,             , committed:             , end:             
2020/06/02 00:12:47.176 INFO  [di_cust_experience_conversation_processor.processors.conversation_processor] (print_stats)     TP(topic='com.expedia.vap.domain.entities.ConversationSegmentReprocess.gcolab', partition=5), read:,             , committed:             , end:             
2020/06/02 00:12:47.176 INFO  [di_cust_experience_conversation_processor.processors.conversation_processor] (print_stats)     TP(topic='di_data_customer_experience_conversation_processor_al6_lab-__assignor-__leader', partition=0), read:,             , committed:             , end:             
2020/06/02 00:12:47.176 INFO  [di_cust_experience_conversation_processor.processors.conversation_processor] (print_stats) 
2020/06/02 00:12:48.591 INFO  [aiokafka.conn] (_do_sasl_handshake) Authenticated as LB6QLE7QI5UKYXFV via PLAIN
2020/06/02 00:12:49.954 INFO  [di_data_customer_experience_conversation_processor_al6] (reprocess_agent) Reprocess agent is not enabled, so sleeping.
2020/06/02 00:12:50.014 INFO  [di_data_customer_experience_conversation_processor_al6] (log_event_details) Topic Name ~ Partition ~ Offset ~ BrokerTimestamp ~ Key: CONVERSATION_RESERVED~3~428491~1590227564.288~1e5cce38-dcc1-4361-80ad-935ce433e84eData: {'event_id': '89281b2e-f3ae-4ea9-b278-ca6c993b9280', 'event_type': 'CONVERSATION_RESERVED', 'conversation_queued': None, 'conversation_claim_flow': {'agent_login_name': 'Perf_Agent', 'conversation_type': 'VOICE'}, 'conversation_unassigned': None, 'conversation_voice_actions': None, 'human_agent_request': None, 'headers': {}, '_avro': {'schema': <avro.schema.RecordSchema object at 0x122f071d0>, 'schema_id': 773}}
2020/06/02 00:12:50.367 ERROR [faust.agents.actor] (log) [^----Agent*: di_cust_exp[.]conversation_agent]: Crashed reason=Exception('hey!!!!!!!')
Traceback (most recent call last):
  File "/Users/anasthana/.virtualenvs/di-cust-experience-conversation-processor/lib/python3.7/site-packages/faust/agents/agent.py", line 647, in _execute_actor
    await coro
  File "/Users/anasthana/di-cust-experience-conversation-processor/di_cust_experience_conversation_processor/processors/conversation_processor.py", line 49, in conversation_agent
    raise Exception("hey!!!!!!!")
Exception: hey!!!!!!!
2020/06/02 00:12:52.258 INFO  [di_cust_experience_conversation_processor.processors.conversation_processor] (print_stats) Rebalancing: False, Rebalances: 1, Events: 23, Messages: 23, Sent: 0, Messages/s 0
2020/06/02 00:12:52.259 INFO  [di_cust_experience_conversation_processor.processors.conversation_processor] (print_stats)     TP(topic='com.expedia.vap.Conversation.gcolab', partition=0), read:,             , committed:             , end:             
2020/06/02 00:12:52.259 INFO  [di_cust_experience_conversation_processor.processors.conversation_processor] (print_stats)     TP(topic='com.expedia.vap.Conversation.gcolab', partition=1), read:,             , committed:             , end:             
2020/06/02 00:12:52.259 INFO  [di_cust_experience_conversation_processor.processors.conversation_processor] (print_stats)     TP(topic='com.expedia.vap.Conversation.gcolab', partition=2), read:,             , committed:             , end:             
2020/06/02 00:12:52.259 INFO  [di_cust_experience_conversation_processor.processors.conversation_processor] (print_stats)     TP(topic='com.expedia.vap.Conversation.gcolab', partition=3), read:,       428491, committed:       428491, end:      1032411
2020/06/02 00:12:52.259 INFO  [di_cust_experience_conversation_processor.processors.conversation_processor] (print_stats)     TP(topic='com.expedia.vap.Conversation.gcolab', partition=4), read:,             , committed:             , end:             
2020/06/02 00:12:52.259 INFO  [di_cust_experience_conversation_processor.processors.conversation_processor] (print_stats)     TP(topic='com.expedia.vap.Conversation.gcolab', partition=5), read:,             , committed:             , end:             
2020/06/02 00:12:52.259 INFO  [di_cust_experience_conversation_processor.processors.conversation_processor] (print_stats)     TP(topic='com.expedia.vap.domain.entities.ConversationSegmentReprocess.gcolab', partition=0), read:,             , committed:             , end:             
2020/06/02 00:12:52.259 INFO  [di_cust_experience_conversation_processor.processors.conversation_processor] (print_stats)     TP(topic='com.expedia.vap.domain.entities.ConversationSegmentReprocess.gcolab', partition=1), read:,             , committed:             , end:             
2020/06/02 00:12:52.260 INFO  [di_cust_experience_conversation_processor.processors.conversation_processor] (print_stats)     TP(topic='com.expedia.vap.domain.entities.ConversationSegmentReprocess.gcolab', partition=2), read:,             , committed:             , end:             
2020/06/02 00:12:52.260 INFO  [di_cust_experience_conversation_processor.processors.conversation_processor] (print_stats)     TP(topic='com.expedia.vap.domain.entities.ConversationSegmentReprocess.gcolab', partition=3), read:,             , committed:             , end:             
2020/06/02 00:12:52.260 INFO  [di_cust_experience_conversation_processor.processors.conversation_processor] (print_stats)     TP(topic='com.expedia.vap.domain.entities.ConversationSegmentReprocess.gcolab', partition=4), read:,             , committed:             , end:             
2020/06/02 00:12:52.260 INFO  [di_cust_experience_conversation_processor.processors.conversation_processor] (print_stats)     TP(topic='com.expedia.vap.domain.entities.ConversationSegmentReprocess.gcolab', partition=5), read:,             , committed:             , end:             
2020/06/02 00:12:52.260 INFO  [di_cust_experience_conversation_processor.processors.conversation_processor] (print_stats)     TP(topic='di_data_customer_experience_conversation_processor_al6_lab-__assignor-__leader', partition=0), read:,             , committed:             , end:             
2020/06/02 00:12:52.260 INFO  [di_cust_experience_conversation_processor.processors.conversation_processor] (print_stats) 
2020/06/02 00:12:55.370 INFO  [mode.supervisors] (log) [^----OneForOneSupervisor: (1@0x1220e1b50)]: Restarting dead <Agent*: di_cust_exp[.]conversation_agent>! Last crash reason: Exception('hey!!!!!!!')

Versions

  • Python version 3.7.4
  • Faust version 1.10.3
  • Operating system macOS 10.13.4
  • Kafka version 2.5
  • RocksDB version (if applicable) N/A

Source: robinhood/faust

View original on GitHubView discussion on GitHub