René's URL Explorer Experiment

{"@context":"https://schema.org","@type":"DiscussionForumPosting","headline":"Long tasks lead to duplicate messages","articleBody":"Hi,\r\n\r\nI'm having an issue where I have a long process (up to 80 minutes) running on a Kubernetes docker instance.  The instance is running Ubuntu 20.04, and I'm using Python 3.8.10.  The docker container runs a python worker script, which runs a subprocess.  The subprocess is multi-threaded and can use all threads during some CPU intensive tasks.\r\n\r\nI'm getting a lot of duplicates (about 5 to 10 duplicates).  This is repeatable and probably due to the intense CPU usage.  What is the correct way to handle this?  Thanks in advance.\r\n\r\nMy pip versions are:\r\n\r\n```\r\ngoogle-api-core               1.30.0\r\ngoogle-api-python-client      2.38.0\r\ngoogle-auth                   1.30.2\r\ngoogle-auth-httplib2          0.1.0\r\ngoogle-auth-oauthlib          0.4.4\r\ngoogle-cloud-core             1.7.0\r\ngoogle-cloud-firestore        2.3.4\r\ngoogle-cloud-pubsub           2.9.0\r\n```\r\n\r\nMy worker code is similar to this:\r\n\r\n```\r\nimport shutil\r\nimport argparse\r\nimport time\r\nimport json\r\nimport subprocess, shlex\r\nfrom pathlib import Path\r\nimport concurrent.futures\r\nfrom google.cloud import pubsub_v1\r\n\r\ndef run_command(cmd_title, cmd):\r\n\t\"\"\"Run command string as subprocess\"\"\"\r\n\ttry:\r\n\t\tcmd_list = shlex.split(cmd)\r\n\t\toutput = subprocess.run(cmd_list,check=True)\r\n\t\tprint (output)\r\n\texcept subprocess.CalledProcessError as e:\r\n\t\tprint ('Error running command ',str(cmd))\r\n\t\tprint (e.output)\r\n\t\tpass\r\n\r\n\treturn\r\n\r\ndef worker(\r\n\tproject_id,\r\n\tsubscription):\r\n\t''' receives messages from a pull subscription'''\r\n\r\n\t#create subscriber\r\n\tsubscriber = pubsub_v1.SubscriberClient()\r\n\tsubscription_path = subscriber.subscription_path(project_id, subscription)\r\n\tflow_control = pubsub_v1.types.FlowControl(\r\n\t\tmax_messages=1, \r\n\t\tmax_duration_per_lease_extension = 60,\r\n\t\tmax_lease_duration = 7200)\r\n\r\n\tdef callback(message):\r\n\t\tprint ('Received message {} of message ID {}'.format(message, message.message_id))\r\n\r\n\t\tcmd = '......'\r\n\t\trun_command('Long Task', cmd)\r\n\r\n\t\t#ack message\r\n\t\tmessage.ack()\r\n\t\tprint('Acknowledged message of message ID {}\\n'.format(message.message_id))\r\n\r\n\tfuture = subscriber.subscribe(subscription_path, callback=callback, flow_control=flow_control)\r\n\tprint('Listening for messages on {}..\\n'.format(subscription_path))\r\n\r\n\t# Wrap subscriber in a 'with' block to automatically call close() when done.\r\n\twith subscriber:\r\n\t\ttry:\r\n\t\t\t# When `timeout` is not set, result() will block indefinitely,\r\n\t\t\t# unless an exception is encountered first.\r\n\t\t\tfuture.result()\r\n\t\texcept:\r\n\t\t\tprint ('Exiting subscriber')\r\n\t\t\tfuture.cancel()\r\n\r\nif __name__ == '__main__':\r\n\tparser = argparse.ArgumentParser(description = 'Worker process')\r\n\tparser.add_argument('-p', '--project_id',default='project_id')\r\n\t\r\n\tparser.add_argument('-s', '--subscription', default='subscription')\r\n\targs = parser.parse_args()\r\n\r\n\t#run worker\r\n\tworker(args.project_id, args.subscription)\r\n```\r\n\r\nThanks in advance,","author":{"url":"https://github.com/jtressle","@type":"Person","name":"jtressle"},"datePublished":"2022-03-01T08:05:06Z","url":"https://github.com/googleapis/python-pubsub/issues/593"}