We have a data pipeline built in Google Cloud Dataflow that consumes messages from a pubsub topic and streams them into BigQuery. In order to test that it works successfully we have some tests that run in a CI pipeline, these tests post messages onto the pubsub topic and verify that the messages are written to BigQuery successfully.
This is the code that posts to the pubsub topic:
from google.cloud import pubsub_v1
def post_messages(project_id, topic_id, rows)
futures = dict()
publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path(
project_id, topic_id
)
def get_callback(f, data):
def callback(f):
try:
futures.pop(data)
except:
print("Please handle {} for {}.".format(f.exception(), data))
return callback
for row in rows:
# When you publish a message, the client returns a future. Data must be a bytestring
# ...
# construct a message in var json_data
# ...
message = json.dumps(json_data).encode("utf-8")
future = publisher.publish(
topic_path,
message
)
futures_key = str(message)
futures[futures_key] = future
future.add_done_callback(get_callback(future, futures_key))
# Wait for all the publish futures to resolve before exiting.
while futures:
time.sleep(1)
When we run this test in our CI pipeline it has started failing intermittently with error
21:38:55: AuthMetadataPluginCallback "<google.auth.transport.grpc.AuthMetadataPlugin object at 0x7f5247407220>" raised exception!
Traceback (most recent call last):
File "/opt/conda/envs/py3/lib/python3.8/site-packages/grpc/_plugin_wrapping.py", line 89, in __call__
self._metadata_plugin(
File "/opt/conda/envs/py3/lib/python3.8/site-packages/google/auth/transport/grpc.py", line 101, in __call__
callback(self._get_authorization_headers(context), None)
File "/opt/conda/envs/py3/lib/python3.8/site-packages/google/auth/transport/grpc.py", line 87, in _get_authorization_headers
self._credentials.before_request(
File "/opt/conda/envs/py3/lib/python3.8/site-packages/google/auth/credentials.py", line 134, in before_request
self.apply(headers)
File "/opt/conda/envs/py3/lib/python3.8/site-packages/google/auth/credentials.py", line 110, in apply
_helpers.from_bytes(token or self.token)
File "/opt/conda/envs/py3/lib/python3.8/site-packages/google/auth/_helpers.py", line 130, in from_bytes
raise ValueError("***0!r*** could not be converted to unicode".format(value))
ValueError: None could not be converted to unicode
Error: The operation was canceled.
Unfortunately this only fails in our CI pipeline, and even then it is failing intermittently (only fails on a small percentage of all CI pipeline runs). If I run the same test locally it succeeds every time. When running in the CI pipeline the code is authenticating as a service account whereas when I run it locally it is authenticating as myself
I know from the error message that it is failing on this code:
if isinstance(result, six.text_type):
return result
else:
raise ValueError("{0!r} could not be converted to unicode".format(value))
https://github.com/googleapis/google-auth-library-python/blob/3c3fbf40b07e090f2be7fac5b304dbf438b5cd6c/google/auth/_helpers.py#L127-L130
which is in a python library from google that we install using pip.
Clearly the expression:
isinstance(result, six.text_type)
is evaluating to False. I put a breakpoint on that code when I ran it locally and discovered that under normal circumstances (i.e. when it works) the value of result is something like this:

That looks like some sort of auth token.
Given the error message:
ValueError: None could not be converted to unicode
it seems that whatever action is being undertaken by the google authentication libraries it is passing None through to the code shown above.
I am at the bounds of my knowledge here. Given this is only failing in a CI pipeline I don't have the opportunity to put a breakpoint in my code and debug it. Given the call stack in the error message this is something to do with authentication.
I'm hoping someone can advise on a course of action.
Can anyone explain a means by which I can discover why None is being passed through to the code that is raising an error?
We had the same error. Finally solved it by using a JSON Web Token for authentication per Google's Quckstart. Like so:
import json
from google.cloud import pubsub_v1
from google.auth import jwt
def post_messages(credentials_path, topic, list_of_messages):
credentials_dict = json.load(open(credentials_path,'r'))
audience = "https://pubsub.googleapis.com/google.pubsub.v1.Publisher"
credentials_ob = jwt.Credentials.from_service_account_info(
credentials_dict, audience=audience
)
publisher = pubsub_v1.PublisherClient(credentials=credentials_ob)
for message_dict in list_of_message_dicts:
message = json.dumps(message_dict, default=str).encode("utf-8")
future = publisher.publish(topic, message)
We also updated our environment but it didn't fix the ValueError until we changed to jwt. Here's the environment in any case:
google-api-core==2.4.0
google-api-python-client==2.36.0
google-auth==2.3.2
google-auth-httplib2==0.1.0
google-auth-oauthlib==0.4.6
google-cloud-core==2.1.0
google-cloud-pubsub==2.9.0
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With