RunTime error occurs when shutting down the streaming pull manager on user callback exception

[kthulsidoss]

@plamut i no longer receive the deadline exceeded error #36 however i dont see the actions being perfomed under
def callback(message)
Can you please help on this

[plamut]

@kthulsidoss Please provide a code sample so that we can take a look, thanks!

[plamut]

From the comment on another issue:

def synchronous_pull_apply_security_targetapplications(project_id, subscription_name):
    from google.cloud import pubsub_v1

    
    subscriber = pubsub_v1.SubscriberClient()
   
    subscription_path = subscriber.subscription_path(
        project_id, subscription_name)

    def callback(message):
        print("Received message: {}".format(message))
        data = message.decode('utf-8')
        format = "%Y-%m-%d %H:%M:%S %Z%z"
        now_utc = datetime.now(timezone('UTC'))
       
        data = json.loads(data)
        print("TDK printing before the steps that needs to be done after the data -->>>> ",
              now_utc.astimezone(timezone('Asia/Kolkata')).strftime(format))
        print("TDK printing the recieved data >>>>>>******************",
              now_utc.astimezone(timezone('Asia/Kolkata')).strftime(format), data)
        
        print("TDK printing the request_type from synchronous pull method*****************->",
              now_utc.astimezone(timezone('Asia/Kolkata')).strftime(format), data['request_type'])
        print("TDK printing the request_id from synchronous pull *****************->",
              now_utc.astimezone(timezone('Asia/Kolkata')).strftime(format), data['request_id'])
        print("TDK printing the gsr_id from synchronous pull *****************->",
              data['gsr_id'], now_utc.astimezone(timezone('Asia/Kolkata')).strftime(format))
        print("TDK printing the Security-Sync_State from synchronous pull method*****************---->",
              now_utc.astimezone(timezone('Asia/Kolkata')).strftime(format), data['Security_Sync_State'])
        print("TDK printing the s_json from synchronous pull method *****************->",
              data['s_json'], now_utc.astimezone(timezone('Asia/Kolkata')).strftime(format))
        ss = SecAdmSecurityService(data['s_json'], data['request_id'])
        st_json_name = ss.process_the_security_json()
        print("!!!!!!!!!!st_json_name!!!!!!!!!!!!", st_json_name)
        try:
            print("&&&&&&&&&sts&&&&&")
            msas = MasterSecadmService(
                data['request_type'], data['request_id'], data['gsr_id'], data['Security_Sync_State'])
            sts, sts_ms = msas.read_status_json(st_json_name)
            print("&&&&&&&&&stswhile reading json&&&&&", sts, sts_ms)
        except KeyError as e:
            print("@@@@@@@@@Exception@@@@@@@", e)
            s_r = security_requests.objects.filter(
                request_id=data['request_id'])
            print("****************INSIDE EXCEPTION*********")
            s_r.update(status='-1: Error', status_message=str(e))
        else:
            s_r = security_requests.objects.filter(request_id=self.request_id)
            s_r.update(status=sts if sts == 'SUCCESS' else sts,
                       status_message=sts_ms)
        print("Received: {}".format(received_message.message.data))
        
        message.ack()

    streaming_pull_future = subscriber.subscribe(
        subscription_path, callback=callback)
    print("Listening for messages on {}..\n".format(subscription_path))

   
    try:
        streaming_pull_future.result(timeout=timeout)
    except:  # noqa
        streaming_pull_future.cancel()

[kthulsidoss]

@plamut are you needing to see anything from the log ?

[plamut]

From the comment:

also i see that the message has been published and it is in the Topic this means that it was not processed by the subscriber. Although i see a line in the log 'Listening for messages on projects/atomic-monument-267912/subscriptions/security_receive..' and this line is after the message.ack().

It seems that timeout is not defined and the script exits. Could you try setting it?
The NameError is not visible, because the try-except block catches it and simply shuts down the streaming pull on any error.

Edit: I can also see the log output, pulling the messages in the background seems to work, but logging needs to be enabled first. I added the following to the top of the script (change the level to something else, e.g. INFO, if DEBUG is too verbose):

import logging

log_format = (
    "%(levelname)-8s [%(asctime)s] %(threadName)-33s "
    "[%(name)s] [%(filename)s:%(lineno)d][%(funcName)s] %(message)s"
)

logging.basicConfig(
    level=logging.DEBUG,
    format=log_format,
)

[kthulsidoss]

now the result of the api call (async call) is set_result can only be called once.

[kthulsidoss]

here is further problematic ones
Received message: Message {
data: b'{"request_type": "Apply_security_to_target_applica...'
attributes: {}
}
Top-level exception occurred in callback while processing a message
Traceback (most recent call last):
File "/opt/secadm/hsbcsecadm/env/lib/python3.6/site-packages/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py", line 77, in _wrap_callback_errors
callback(message)
File "./api/views.py", line 322, in callback
data = message.decode('utf-8')

[plamut]

@kthulsidoss I was able to reproduce it, and the following appears in the log output:

    ...
    data = message.decode('utf-8')
AttributeError: 'Message' object has no attribute 'decode'

Messages received with the streaming pull are wrapped into a Message class, thus the following:

data = message.decode('utf-8')

needs to be changed to:

data = message.data.decode('utf-8')

With that said, the following should still not occur during the shutdown if the user callback raises an exception:

RuntimeError: set_result can only be called once.

[plamut]

I changed the issue title to reflect the issue on shutdown that was revealed here (the original issue was caused due to the bug in the provided callback).