Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Optionaly return schema id together with decoded message #3

Open
wants to merge 3 commits into
base: master
Choose a base branch
from
Open

Optionaly return schema id together with decoded message #3

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
bd13c2a
optionaly return schema id together with decoded message
michaelr524 Feb 16, 2016
ad46a27
adding lame printf debugging for schema_id error
afsheenb Apr 28, 2016
d7a525c
remove schema_id reference
afsheenb Apr 28, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 8 additions & 2 deletions confluent/schemaregistry/serializers/MessageSerializer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,6 +55,8 @@ def encode_record_with_schema(self, topic, schema, record, is_key=False):
try:
schema_id = self.registry_client.register(subject, schema)
except:
message_none = "Exception, schema_id will be set to None!"
raise SerializerError(message_none)
schema_id = None

if not schema_id:
Expand Down Expand Up @@ -166,7 +168,7 @@ def decoder(p):
self.id_to_decoder_func[schema_id] = decoder
return self.id_to_decoder_func[schema_id]

def decode_message(self, message):
def decode_message(self, message, with_schema_id=False):
"""
Decode a message from kafka that has been encoded for use with
the schema registry.
Expand All @@ -179,4 +181,8 @@ def decode_message(self, message):
if magic != MAGIC_BYTE:
raise SerializerError("message does not start with magic byte")
decoder_func = self._get_decoder_func(schema_id, payload)
return decoder_func(payload)
decoded = decoder_func(payload)
if with_schema_id:
return (decoded, schema_id)
else:
return decoded