feat: Kafka Protocol #197

davidwmartines · 2022-11-09T05:53:55Z

Signed-off-by: davidwmartines d5172@yahoo.com

Fixes #196

Changes

This PR adds functions for converting CloudEvents to Kakfa-protocol-ready objects, and for converting Kafka-protocol messages into CloudEvents, per the CloudEvents Kafka protocol binding spec.

Supports binary or structured mode conversions
Clients can supply appropriate serialization and deserializer functions as MarshallerType callables.

Add Kafka Protocol support

Tests pass
Appropriate changes to README are included in PR

Signed-off-by: davidwmartines <d5172@yahoo.com>

for more information, see https://pre-commit.ci

Signed-off-by: davidwmartines <d5172@yahoo.com>

…s/sdk-python into add-kafka-protocol Signed-off-by: davidwmartines <d5172@yahoo.com>

for more information, see https://pre-commit.ci

Signed-off-by: davidwmartines <d5172@yahoo.com>

…s/sdk-python into add-kafka-protocol

xSAVIKx

@davidwmartines thank you for your PR! This looks really good to me.

Can you please take a look at the comments posted below? I believe there is a bit of a misunderstanding between the spec and the implementation (or I may have misunderstood the spec). It'd be great to align the implementation with the spec or raise some questions regarding how the spec should translate into implementation. As far as I see, all the other implementations use partitionkey over key in the CloudEvent attributes (https://github.com/search?q=org%3Acloudevents+partitionkey&type=code)

xSAVIKx · 2022-11-11T08:45:21Z

cloudevents/kafka/conversion.py

+    Returns a CloudEvent from a Kafka ProtocolMessage in binary format.
+
+    :param message: The ProtocolMessage to be converted.
+    :param event_type:  The type of CloudEvent to create.  Defaults to http.CloudEvent.


Suggested change

:param event_type: The type of CloudEvent to create. Defaults to http.CloudEvent.

:param event_type: The type of CloudEvent to create. Defaults to http.CloudEvent.

xSAVIKx · 2022-11-11T08:46:38Z

cloudevents/kafka/conversion.py

+    headers: typing.Dict[str, bytes]
+    key: typing.Union[bytes, str]
+    value: typing.Union[bytes, str]


Can you please add missing docs?

xSAVIKx · 2022-11-11T08:48:04Z

cloudevents/kafka/conversion.py

+DEFAULT_EMBEDDED_DATA_MARSHALLER = lambda x: x
+
+
+class ProtocolMessage(typing.NamedTuple):


I'm not sure from where this name comes. Is it some standard conventional name for Kafka?

I've glanced at the spec, looks like in Kafka they have either a Message or a Record. Maybe rename this to KafkaMessage instead?

xSAVIKx · 2022-11-11T08:48:27Z

cloudevents/kafka/conversion.py

+
+
+def to_binary(
+    event: AnyCloudEvent, data_marshaller: types.MarshallerType = None


Suggested change

event: AnyCloudEvent, data_marshaller: types.MarshallerType = None

event: AnyCloudEvent, data_marshaller: types.Optional[types.MarshallerType] = None

xSAVIKx · 2022-11-11T08:51:19Z

cloudevents/kafka/conversion.py

+    if isinstance(data, str):
+        data = data.encode("utf-8")
+
+    return ProtocolMessage(headers, event.get("key"), data)


So as stated in the Kafka spec, we should provide a way for a user to pass a mapper function for getting the key.

I'd very much like to have that option and have a default that's gonna just use .get("key"). Also, while the key is a required parameter, maybe we can move this key mapping functionality to the very top of the method and fail fast with an error if it is not possible to get a key?

xSAVIKx · 2022-11-11T08:59:51Z

cloudevents/kafka/conversion.py

+    """
+
+    headers: typing.Dict[str, bytes]
+    key: typing.Union[bytes, str]


As far as I see from the spec, the name of this attribute should be partitionkey in the CloudEvent.

Every implementation SHOULD provide an opt-in "Key Mapper" implementation that maps the [Partitioning](https://github.com/cloudevents/spec/blob/main/cloudevents/extensions/partitioning.md) partitionkey attribute value to the 'key' of the Kafka message as-is, if present.

xSAVIKx · 2022-11-11T09:00:53Z

cloudevents/tests/test_kafka_conversions.py

+
+@pytest.fixture
+def bad_marshaller() -> types.MarshallerType:
+    return _throw


why not just raise an error here?

cloudevents/tests/test_kafka_conversions.py

xSAVIKx · 2022-11-11T09:01:59Z

cloudevents/tests/test_kafka_conversions.py

+def test_throw_raises():
+    with pytest.raises(Exception):
+        _throw()


that's definitely should not be here. Why write a test for a method from a test?

xSAVIKx · 2022-11-11T09:03:01Z

cloudevents/tests/test_kafka_conversions.py

+def test_to_structured_none_key(source_event):
+    source_event["key"] = None
+    result = to_structured(source_event)
+    assert result.key is None
+
+
+def test_to_structured_no_key(source_event):
+    del source_event["key"]
+    result = to_structured(source_event)
+    assert result.key is None


I believe these tests should actually assert that an error is raised when no partitionkey is available in the event or the mapping function fails to do it's job.

davidwmartines · 2022-11-11T13:24:58Z

@xSAVIKx Thanks for the great feedback. I will get to work on this!

I agree on all points, except for requiring the key. Kafka does allow for null message keys, so I think we can allow a key of None. Let me know if you think this needs more discussion. I will try to clarify this when I add docs to the KafkaMessage class.

Good call on the key mapper function - I was wondering about that. I will add the option to supply a key-mapper as well.

Signed-off-by: davidwmartines <d5172@yahoo.com>

davidwmartines · 2022-11-11T22:53:01Z

working on key_mapper next...

Signed-off-by: davidwmartines <d5172@yahoo.com>

xSAVIKx

@davidwmartines overall LGTM! thanks again for you work. Can you please address these 2 minor comments before we merge the PR?

xSAVIKx · 2022-11-14T19:57:50Z

cloudevents/kafka/conversion.py

+    """
+
+
+KeyMapper = typing.Optional[typing.Callable[[AnyCloudEvent], typing.Union[bytes, str]]]


Suggested change

KeyMapper = typing.Optional[typing.Callable[[AnyCloudEvent], typing.Union[bytes, str]]]

KeyMapper = typing.Callable[[AnyCloudEvent], typing.Union[bytes, str]]

KeyMapper itself is not an optional function. But it is not required, so IMO Optional belongs to the function declaration, not the type alias.

xSAVIKx · 2022-11-14T20:00:04Z

cloudevents/kafka/exceptions.py

@@ -0,0 +1,7 @@
+from cloudevents import exceptions as cloud_exceptions


Can you please add a copyright header to this file?

Signed-off-by: davidwmartines <d5172@yahoo.com>

davidwmartines · 2022-11-14T21:44:37Z

@davidwmartines overall LGTM! thanks again for you work. Can you please address these 2 minor comments before we merge the PR?

Thanks!

davidwmartines and others added 2 commits November 8, 2022 23:45

Add kafka event and conversions.

c4e726b

Signed-off-by: davidwmartines <d5172@yahoo.com>

[pre-commit.ci] auto fixes from pre-commit.com hooks

ca8d877

for more information, see https://pre-commit.ci

davidwmartines changed the title ~~Add kafka event and conversions.~~ Kafka Protocol Nov 9, 2022

This comment was marked as resolved.

Sign in to view

davidwmartines and others added 9 commits November 9, 2022 09:48

Remove kafka CloudEvent class

086b135

Signed-off-by: davidwmartines <d5172@yahoo.com>

Update conversion and init

ef71c6e

Signed-off-by: davidwmartines <d5172@yahoo.com>

Merge branch 'add-kafka-protocol' of https://github.com/davidwmartine…

070f1ab

…s/sdk-python into add-kafka-protocol Signed-off-by: davidwmartines <d5172@yahoo.com>

[pre-commit.ci] auto fixes from pre-commit.com hooks

72fb800

for more information, see https://pre-commit.ci

Fix formatting.

714c159

Signed-off-by: davidwmartines <d5172@yahoo.com>

Add tests for kafka binary conversion.

4bd1405

Signed-off-by: davidwmartines <d5172@yahoo.com>

Catch marshalling errors, raise cloud_exceptions.

1af9b22

Signed-off-by: davidwmartines <d5172@yahoo.com>

Add tests for to/from structured.

cb6c0a2

Signed-off-by: davidwmartines <d5172@yahoo.com>

Merge branch 'add-kafka-protocol' of https://github.com/davidwmartine…

6d63e47

…s/sdk-python into add-kafka-protocol

davidwmartines changed the title ~~Kafka Protocol~~ feat: Kafka Protocol Nov 10, 2022

davidwmartines marked this pull request as ready for review November 10, 2022 23:12

xSAVIKx self-requested a review November 11, 2022 08:35

xSAVIKx reviewed Nov 11, 2022

View reviewed changes

davidwmartines added 7 commits November 11, 2022 15:33

Fix spacing issues.

0d58b97

Signed-off-by: davidwmartines <d5172@yahoo.com>

Rename ProtocolMessage to KafkaMessage.

5d84392

Signed-off-by: davidwmartines <d5172@yahoo.com>

Correct type annotations.

dd21f16

Signed-off-by: davidwmartines <d5172@yahoo.com>

Use .create function.

6b7c373

Signed-off-by: davidwmartines <d5172@yahoo.com>

Simplify failing serdes function.

b1cee7e

Signed-off-by: davidwmartines <d5172@yahoo.com>

Organize tests into classes.

bfd17cc

Signed-off-by: davidwmartines <d5172@yahoo.com>

Fix partitionkey attribute name and logic.

1a6cecb

Signed-off-by: davidwmartines <d5172@yahoo.com>

davidwmartines added 2 commits November 13, 2022 09:19

Add key_mapper option.

eea5ae1

Signed-off-by: davidwmartines <d5172@yahoo.com>

Refactor tests, raise KeyMapperError

acebd72

Signed-off-by: davidwmartines <d5172@yahoo.com>

davidwmartines requested a review from xSAVIKx November 13, 2022 16:04

xSAVIKx approved these changes Nov 14, 2022

View reviewed changes

xSAVIKx added the enhancement New feature or request label Nov 14, 2022

davidwmartines added 2 commits November 14, 2022 15:38

Add copyright.x

d5921a9

Signed-off-by: davidwmartines <d5172@yahoo.com>

Remove optional typing.

31e8fbc

Signed-off-by: davidwmartines <d5172@yahoo.com>

davidwmartines requested a review from xSAVIKx November 14, 2022 21:44

xSAVIKx approved these changes Nov 17, 2022

View reviewed changes

xSAVIKx merged commit de61dd9 into cloudevents:main Nov 17, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: Kafka Protocol #197

feat: Kafka Protocol #197

davidwmartines commented Nov 9, 2022 •

edited

Loading

This comment was marked as resolved.

xSAVIKx left a comment •

edited

Loading

xSAVIKx Nov 11, 2022

xSAVIKx Nov 11, 2022

xSAVIKx Nov 11, 2022

xSAVIKx Nov 11, 2022

xSAVIKx Nov 11, 2022

xSAVIKx Nov 11, 2022

xSAVIKx Nov 11, 2022

xSAVIKx Nov 11, 2022

xSAVIKx Nov 11, 2022

davidwmartines commented Nov 11, 2022

davidwmartines commented Nov 11, 2022

xSAVIKx left a comment

xSAVIKx Nov 14, 2022

xSAVIKx Nov 14, 2022

davidwmartines commented Nov 14, 2022

	:param event_type: The type of CloudEvent to create. Defaults to http.CloudEvent.
	:param event_type: The type of CloudEvent to create. Defaults to http.CloudEvent.

		DEFAULT_EMBEDDED_DATA_MARSHALLER = lambda x: x


		class ProtocolMessage(typing.NamedTuple):



		def to_binary(
		event: AnyCloudEvent, data_marshaller: types.MarshallerType = None

		"""


		KeyMapper = typing.Optional[typing.Callable[[AnyCloudEvent], typing.Union[bytes, str]]]

	KeyMapper = typing.Optional[typing.Callable[[AnyCloudEvent], typing.Union[bytes, str]]]
	KeyMapper = typing.Callable[[AnyCloudEvent], typing.Union[bytes, str]]

		@@ -0,0 +1,7 @@
		from cloudevents import exceptions as cloud_exceptions

feat: Kafka Protocol #197

feat: Kafka Protocol #197

Conversation

davidwmartines commented Nov 9, 2022 • edited Loading

Changes

Add Kafka Protocol support

This comment was marked as resolved.

xSAVIKx left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

davidwmartines commented Nov 11, 2022

davidwmartines commented Nov 11, 2022

xSAVIKx left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

davidwmartines commented Nov 14, 2022

davidwmartines commented Nov 9, 2022 •

edited

Loading

xSAVIKx left a comment •

edited

Loading