add kafka topic/partition/offset to the extension of event #896

yanmxa · 2023-05-12T04:15:05Z

Signed-off-by: myan [email protected]
Refers: #846

embano1 · 2023-05-12T06:28:00Z

Can you please update the PR to include topic?

nit: you could slightly optimize this existing call to avoid additional allocations if current headers is 0:

headers := make(map[string][]byte, 3+len(cm.Headers))

Signed-off-by: myan <[email protected]>

yanmxa · 2023-05-14T13:55:00Z

Can you please update the PR to include topic?

nit: you could slightly optimize this existing call to avoid additional allocations if current headers is 0:
headers := make(map[string][]byte, 3+len(cm.Headers))

Thanks @embano1 ! The kafka topic has been added the in this pr and also optimization of the header initialization is done.

embano1

lgtm cc/ @duglin

duglin · 2023-05-15T15:06:57Z

@yanmxa can you look at the test failures?

duglin · 2023-05-15T15:07:18Z

it might also be good to add some new ones

Signed-off-by: myan <[email protected]>

yanmxa · 2023-05-16T01:43:09Z

@yanmxa can you look at the test failures?

@duglin Sure, I added few codes for the integration test.

embano1 · 2023-05-16T06:25:51Z

Error:      	Received unexpected error:
        	            	kafka server: In the middle of a leadership election, there is currently no leader for this partition and hence it is unavailable for writes

Probably not related to this PR, but worth taking a look?

duglin · 2023-05-16T12:52:11Z

I'm going to kick off the tests again, but here's a link to the current test run for reference: https://github.com/cloudevents/sdk-go/actions/runs/4986878143/jobs/8928440318?pr=896
hopefully it'll still work after I rerun stuff

duglin · 2023-05-16T13:17:21Z

Still seeming failures due to the new fields:
https://github.com/cloudevents/sdk-go/actions/runs/4986878143/jobs/8939427606?pr=896

Signed-off-by: myan <[email protected]>

yanmxa · 2023-05-17T03:21:49Z

From the error message, it is caused by the Sarama binding integration test, and I add some code to fix the error.

duglin · 2023-05-17T12:42:03Z

@embano1 @lionelvillard for another round of reviews.
LGTM

yanmxa · 2023-05-23T06:57:32Z

Thanks @duglin @embano1 @lionelvillard! PTAL~

embano1 · 2023-05-23T07:13:16Z

test/integration/kafka_sarama/kafka_test.go

+			eventIn.SetExtension(KAFKA_OFFSET, got.Extensions()[KAFKA_OFFSET])
+			eventIn.SetExtension(KAFKA_PARTITION, got.Extensions()[KAFKA_PARTITION])
+			eventIn.SetExtension(KAFKA_TOPIC, got.Extensions()[KAFKA_TOPIC])


Haven't looked deeper at the different test.<> functions, but was wondering what this is supposed to do here? You're setting extensions on the input event based on got and then asserting. Is this correctly exercising and asserting the changes here?

There is another way we can remove the KAFKA_OFFSET , KAFKA_PARTITION, and KAFKA_TOPIC from got. Is this way better?

I'm curious, how do we remove extension attributes from a CE? I didn't see that option in here: https://github.com/cloudevents/sdk-go/blob/main/v2/event/event_writer.go

I'm ok with copying things from got into eventIn so that the Assert works (or removing them from got if that's possible), but it seems to me that there should be a check of the values in got to make sure they're not blank - even if we need to do those manually instead via a compare with eventIn.

@duglin Thanks for your advice! I have added the empty check for the event extensions.

Signed-off-by: myan <[email protected]>

duglin · 2023-05-23T17:21:25Z

test/integration/kafka_sarama/kafka_test.go

+			extensions := got.Extensions()
+			require.NotEmpty(t, extensions[KAFKA_OFFSET])
+			require.NotEmpty(t, extensions[KAFKA_PARTITION])
+			require.NotEmpty(t, extensions[KAFKA_TOPIC])


can we check their values? Should they match the constants defined above?

Sure! But we can only verify the topic defined above. And the partition/offset isn't defined here.

Signed-off-by: myan <[email protected]>

embano1 · 2023-05-24T06:14:28Z

test/integration/kafka_sarama/kafka_test.go

 func TestSendEvent(t *testing.T) {
 	test.EachEvent(t, test.Events(), func(t *testing.T, eventIn event.Event) {
 		eventIn = test.ConvertEventExtensionsToString(t, eventIn)
 		clienttest.SendReceive(t, func() interface{} {
 			return protocolFactory(t)
 		}, eventIn, func(e event.Event) {
-			test.AssertEventEquals(t, eventIn, test.ConvertEventExtensionsToString(t, e))
+			got := test.ConvertEventExtensionsToString(t, e)
+
+			require.Equal(t, TopicName, got.Extensions()[KAFKA_TOPIC])
+			require.NotNil(t, got.Extensions()[KAFKA_PARTITION])
+			require.NotNil(t, got.Extensions()[KAFKA_OFFSET])
+
+			eventIn.SetExtension(KAFKA_OFFSET, got.Extensions()[KAFKA_OFFSET])
+			eventIn.SetExtension(KAFKA_PARTITION, got.Extensions()[KAFKA_PARTITION])
+			eventIn.SetExtension(KAFKA_TOPIC, TopicName)
+			test.AssertEventEquals(t, eventIn, got)
 		})
 	})
 }


Suggesting the following changes for more clarity of the test (also for the other integration test):

func TestSendEvent(t *testing.T) { test.EachEvent(t, test.Events(), func(t *testing.T, eventIn event.Event) { eventIn = test.ConvertEventExtensionsToString(t, eventIn) clienttest.SendReceive(t, func() interface{} { return protocolFactory(t) }, eventIn, func(eventOut event.Event) { eventOut = test.ConvertEventExtensionsToString(t, eventOut) require.Equal(t, TopicName, eventOut.Extensions()[KAFKA_TOPIC]) require.NotNil(t, eventOut.Extensions()[KAFKA_PARTITION]) require.NotNil(t, eventOut.Extensions()[KAFKA_OFFSET]) test.AllOf( test.HasExactlyAttributesEqualTo(eventIn.Context), test.HasData(eventIn.Data()), ) }) }) }

Renamed the variable names for more clarity and dropped extension comparison since we're not setting specific extensions here and have require... to assert on the specific extensions we want. Ideally we'd have a custom test.EventMatcher to assert on extensions but skip specific ones which get added at a later stage. I'll create tracking issue.

Done!
Thank @embano1!

Signed-off-by: myan <[email protected]>

duglin · 2023-05-24T15:02:32Z

@embano1 you good with the latest?

embano1 · 2023-05-24T15:04:50Z

LGTM, thank you!

duglin · 2023-05-24T15:12:40Z

@yanmxa thanks for your patience.
LGTM
ping @lionelvillard for the final review/merge...

lionelvillard · 2023-05-25T15:46:57Z

LGTM

add kafka partition/offset to the extension of event

2a574e8

Signed-off-by: myan <[email protected]>

initialize header with kafka information

455fd8f

Signed-off-by: myan <[email protected]>

yanmxa changed the title ~~add kafka partition/offset to the extension of event~~ add kafka topic/partition/offset to the extension of event May 14, 2023

embano1 approved these changes May 14, 2023

View reviewed changes

fixed the integration test

e033080

Signed-off-by: myan <[email protected]>

fixed the sarama binding test issue

1ab33b9

Signed-off-by: myan <[email protected]>

embano1 reviewed May 23, 2023

View reviewed changes

add check empty in the integration test

22ddc6c

Signed-off-by: myan <[email protected]>

yanmxa force-pushed the br_kafka_offset branch from 69ea310 to 22ddc6c Compare May 23, 2023 16:49

duglin reviewed May 23, 2023

View reviewed changes

verfiy the kafka topic value

cfac14f

Signed-off-by: myan <[email protected]>

embano1 reviewed May 24, 2023

View reviewed changes

clarify the test code

998a3a0

Signed-off-by: myan <[email protected]>

lionelvillard merged commit b7a65db into cloudevents:main May 25, 2023

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

add kafka topic/partition/offset to the extension of event #896

add kafka topic/partition/offset to the extension of event #896

yanmxa commented May 12, 2023

embano1 commented May 12, 2023

yanmxa commented May 14, 2023 •

edited

Loading

embano1 left a comment •

edited

Loading

duglin commented May 15, 2023

duglin commented May 15, 2023

yanmxa commented May 16, 2023 •

edited

Loading

embano1 commented May 16, 2023

duglin commented May 16, 2023

duglin commented May 16, 2023

yanmxa commented May 17, 2023 •

edited

Loading

duglin commented May 17, 2023

yanmxa commented May 23, 2023 •

edited

Loading

embano1 May 23, 2023

yanmxa May 23, 2023

duglin May 23, 2023

yanmxa May 23, 2023

duglin May 23, 2023

yanmxa May 24, 2023

embano1 May 24, 2023

embano1 May 24, 2023

yanmxa May 24, 2023

duglin commented May 24, 2023

embano1 commented May 24, 2023

duglin commented May 24, 2023

lionelvillard commented May 25, 2023

add kafka topic/partition/offset to the extension of event #896

add kafka topic/partition/offset to the extension of event #896

Conversation

yanmxa commented May 12, 2023

embano1 commented May 12, 2023

yanmxa commented May 14, 2023 • edited Loading

embano1 left a comment • edited Loading

Choose a reason for hiding this comment

duglin commented May 15, 2023

duglin commented May 15, 2023

yanmxa commented May 16, 2023 • edited Loading

embano1 commented May 16, 2023

duglin commented May 16, 2023

duglin commented May 16, 2023

yanmxa commented May 17, 2023 • edited Loading

duglin commented May 17, 2023

yanmxa commented May 23, 2023 • 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

duglin commented May 24, 2023

embano1 commented May 24, 2023

duglin commented May 24, 2023

lionelvillard commented May 25, 2023

yanmxa commented May 14, 2023 •

edited

Loading

embano1 left a comment •

edited

Loading

yanmxa commented May 16, 2023 •

edited

Loading

yanmxa commented May 17, 2023 •

edited

Loading

yanmxa commented May 23, 2023 •

edited

Loading