René's URL Explorer Experiment

{"@context":"https://schema.org","@type":"DiscussionForumPosting","headline":"Remove expired ack_ids from lease management.","articleBody":"Possible root cause of: https://github.com/googleapis/python-pubsub/issues/593\r\n\r\n#### Steps to reproduce\r\n\r\n  1. High-latency subscriber (e.g. 2 hours)\r\n  2. Low message count flow control\r\n  3. Solution only works for exactly once delivery subscriptions, since they return invalid_ack_id errors.\r\n\r\n#### Code example\r\n\r\n```\r\n\r\ndef f(x):\r\n    set_time = 60\r\n    timeout = time.time() + 60*float(set_time)  # X minutes from now\r\n    print(timeout)\r\n    while True:\r\n        if time.time() > timeout:\r\n            print(\"Time to ack!\")\r\n            print(time.time())\r\n            break\r\n        x * x\r\n\r\ndef create_load(percent_load):\r\n    processes = psutil.cpu_count()\r\n    to_use = int(processes * (percent_load/100))\r\n    print ('utilizing %d cores\\n' % to_use)\r\n    pool = Pool(to_use)\r\n    pool.map(f, range(to_use))\r\n\r\ndef receive_messages_with_flow_control() -> None:\r\n    \"\"\"Receives messages from a pull subscription with flow control.\"\"\"\r\n    from concurrent.futures import TimeoutError\r\n    from google.cloud import pubsub_v1\r\n\r\n    subscriber = pubsub_v1.SubscriberClient()\r\n    subscription_path = subscriber.subscription_path(project_id, subscription_id)\r\n\r\n    def callback(message: pubsub_v1.subscriber.message.Message) -> None:\r\n        print(f\"Received {message.data!r} at {time.time()}.\")\r\n        create_load(50)\r\n        message.ack()\r\n\r\n    # Limit the subscriber to only have ten outstanding messages at a time.\r\n    flow_control = pubsub_v1.types.FlowControl(max_messages=1)\r\n\r\n    streaming_pull_future = subscriber.subscribe(\r\n        subscription_path, callback=callback, flow_control=flow_control\r\n    )\r\n\r\n    # Wrap subscriber in a 'with' block to automatically call close() when done.\r\n    with subscriber:\r\n        try:\r\n            # When `timeout` is not set, result() will block indefinitely,\r\n            # unless an exception is encountered first.\r\n            streaming_pull_future.result()\r\n        except TimeoutError:\r\n            streaming_pull_future.cancel()  # Trigger the shutdown.\r\n            streaming_pull_future.result()  # Block until the shutdown is complete.\r\n```\r\n\r\n### Error trace:\r\n```\r\ngoogle.api_core.exceptions.InvalidArgument: 400 Some acknowledgement ids in the request were invalid. This could be because the acknowledgement ids have expired or the acknowledgement ids were malformed. [reason: \"EXACTLY_ONCE_ACKID_FAILURE\"\r\ndomain: \"pubsub.googleapis.com\"\r\nmetadata {\r\n  value: \"PERMANENT_FAILURE_INVALID_ACK_ID\"\r\n}\r\n```\r\n\r\nEven while flow control is activate and CPS client library does not receive messages in streaming_pull_manager._on_response, CPS still delivers messages to the bidi stream, which are buffered. These messages are not lease managed (because the library does not know about them) , so they expire before they are ever sent to a callback.\r\n\r\nWhen one of these expired messages is passed to _on_response, and a receipt modack is sent via send_lease_modacks, if we receive an EOD related INVALID_ACK_ID failure, we should not pass these messages on to be processed, as they are already expired.\r\n","author":{"url":"https://github.com/acocuzzo","@type":"Person","name":"acocuzzo"},"datePublished":"2022-09-21T19:54:12Z","url":"https://github.com/googleapis/python-pubsub/issues/786"}