Confluent kafka binding #988

yanmxa · 2023-12-07T10:26:20Z

Adding the Kafka binding with https://github.com/confluentinc/confluent-kafka-go. This implementation will support the following features compared to the sarama binding.

The binding provides the ability to submit offset asynchronously. ref: How to submit message offset asynchronously based on kafka protocol #846
Be able to subscribe to batch topics with a regular pattern. ref: Pattern subscription confluentinc/confluent-kafka-go#96
Support to consume the event from a specific position(topic-partition-offset)
Support to invoke the customized rebalance callback function for the consumer group

Note: In the confluent samples, with configuration WithPollGoroutines(1) to ensure the order of the messages on the partition.

Since Kafka already natively utilizes consumer groups to enhance consumption efficiency, there is no need to use multiple goroutines to boost consumption speed.

Resolve: #918

yanmxa · 2023-12-12T01:06:55Z

/cc @embano1 @duglin @lionelvillard @clyang82

test/integration/kafka_confluent/kafka_test.go

v2/context/context.go

yanmxa · 2024-01-08T04:02:25Z

@embano1 Thanks for your review!

embano1 · 2024-01-23T21:29:42Z

Thy @yanmxa and sorry for my slow responses. I'm under the water at the moment at work and can't give you a concrete date when I'll be able to review. Perhaps @duglin and @lionelvillard can take another look?

lionelvillard · 2024-01-24T13:36:54Z

@yanmxa can you update the documentation: https://github.com/cloudevents/sdk-go/blob/main/docs/protocol_implementations.md?

Also I wonder if it makes sense to indicate (maybe in the documentation) this is a new implementation and it hasn't been widely tested in the wild. WDYT?

yanmxa · 2024-01-26T02:44:55Z

@yanmxa can you update the documentation: https://github.com/cloudevents/sdk-go/blob/main/docs/protocol_implementations.md?

Also I wonder if it makes sense to indicate (maybe in the documentation) this is a new implementation and it hasn't been widely tested in the wild. WDYT?

Thanks @lionelvillard! I update the protocol binding of the related document, #1008. PTAL~

yanmxa · 2024-01-30T15:30:15Z

Thanks @duglin !

yanmxa · 2024-03-06T07:02:46Z

@duglin @embano1 @lionelvillard PTAL~

embano1

I haven't checked the whole PR because for some parts especially structured/binary handling, I'd like to see more tests for non-happy path scenarios and mis-use if possible. Could you please also add Confluent to the Github integration tests so we run this against Confluent images?

protocol/kafka_confluent/v2/go.mod

protocol/kafka_confluent/v2/message.go

embano1 · 2024-03-10T07:08:31Z

protocol/kafka_confluent/v2/message.go

+			contentType = string(header.Value)
+		}
+		if k == specs.PrefixedSpecVersionName() {
+			contentVersion = string(header.Value)


question: what if header.Value is 1.0.2 but MUST be 1.0 according to

Compliant event producers MUST use a value of 1.0 when referring to this version of the specification
https://github.com/cloudevents/spec/blob/v1.0.2/cloudevents/spec.md#specversion

I‘m not sure how to handle the specversion with value 1.0.2. Maybe just validate the 'major' and 'minor' versions whether compliant with "1.0"? and if not, then print a warning message?

I checked the other protocol. Then all just assert whether the specversion key exists and ignore the value.

Do you have any other advice?

Yeah, IMHO this is a shortcoming in the other implementations. I'm fine leaving this here as is then given it's consistent with our other implementations. cc/ @duglin

I like being forgiving and just ignoring the patch version number - if we want to do anything different at all. I'm ok with the current choice of only looking for 0.3 and 1.0.

protocol/kafka_confluent/v2/message.go

protocol/kafka_confluent/v2/option.go

yanmxa · 2024-03-11T10:35:06Z

I haven't checked the whole PR because for some parts especially structured/binary handling, I'd like to see more tests for non-happy path scenarios and mis-use if possible. Could you please also add Confluent to the Github integration tests so we run this against Confluent images?

Thank @embano1!

I added some tests for the misuse cases when trying to initialize the client. If others come up with scenarios beyond these, we can continually improve it in the future.

Also, I updated the GitHub integration test workflow to add the kafka_confluent service(running on port 9192, the sarama running on 9092) with the confluent image. So the current confluent kafka protocol uses it for the tests.

.github/workflows/integration.yaml

protocol/kafka_confluent/v2/message.go

protocol/kafka_confluent/v2/message_test.go

protocol/kafka_confluent/v2/option.go

samples/kafka_confluent/go.mod

protocol/kafka_confluent/v2/message.go

embano1 · 2024-03-25T10:43:08Z

I have a slight concern, perhaps unnecessary and we can just ignore it. That is what if the producer isn't closed by the Closer, that will cause producerDeliveryChan not to be closed?

Please add a comment to the Close(ctx) method that protocol users must call Close(ctx) to properly clean up resources after use. Also, add defer p.Close(ctx) to all your examples

yanmxa · 2024-03-25T10:54:41Z

I have a slight concern, perhaps unnecessary and we can just ignore it. That is what if the producer isn't closed by the Closer, that will cause producerDeliveryChan not to be closed?

Please add a comment to the Close(ctx) method that protocol users must call Close(ctx) to properly clean up resources after use. Also, add defer p.Close(ctx) to all your examples

Done!

embano1 · 2024-03-25T11:29:55Z

Please add a doc string to the Close() method as well, advising users to call Close() after use. Please directly squash your commits so we can merge. Also, any related documentation/README updates needed to further highlight this new protocol implementation?

embano1 · 2024-03-25T11:31:57Z

Also, why did you not use WithPollGoroutines in all receiver examples? Should be clear why this is advised e.g., also adding a comment to the example explaining that this won't have perf impact and is only required if the user cares about in-order processing when using the SDK.

yanmxa · 2024-03-25T13:32:42Z

Also, why did you not use WithPollGoroutines in all receiver examples? Should be clear why this is advised e.g., also adding a comment to the example explaining that this won't have perf impact and is only required if the user cares about in-order processing when using the SDK.

@embano1 Thanks for your prompt feedback!

Add some comments for the Close method and WithPollGoroutines in all receiver examples.
Update the related document in this repository.
Squash the commits

Could you please help me to check if the wording in the above document is appropriate?

protocol/kafka_confluent/v2/protocol.go

embano1

Just that one comment please

embano1 · 2024-03-25T20:51:58Z

Can you please check your integration tests? They're not looking positive, but passing: https://github.com/cloudevents/sdk-go/actions/runs/8420863202/job/23056453953?pr=988#step:5:284 (example)

embano1 · 2024-03-25T20:55:11Z

This unit test is also not looking ok: https://github.com/cloudevents/sdk-go/actions/runs/8420863203/job/23056453818#step:4:3883

yanmxa · 2024-03-26T02:56:51Z

This unit test is also not looking ok: https://github.com/cloudevents/sdk-go/actions/runs/8420863203/job/23056453818#step:4:3883

The error log disappears in the unit test now.

yanmxa · 2024-03-26T03:28:12Z

Can you please check your integration tests? They're not looking positive, but passing: https://github.com/cloudevents/sdk-go/actions/runs/8420863202/job/23056453953?pr=988#step:5:284 (example)

Done!
The integration test error log is from here.

Signed-off-by: myan <myan@redhat.com> add message implementation Signed-off-by: myan <myan@redhat.com> add ut test Signed-off-by: myan <myan@redhat.com> add integration test Signed-off-by: myan <myan@redhat.com> add integration test and samples Signed-off-by: Meng Yan <myan@redhat.com> offset Signed-off-by: Meng Yan <myan@redhat.com> remove the ctx Signed-off-by: myan <myan@redhat.com> review Signed-off-by: Meng Yan <myan@redhat.com> remove Signed-off-by: Meng Yan <myan@redhat.com> init consumer and producer on 1 client Signed-off-by: myan <myan@redhat.com> reply the reviews Signed-off-by: myan <myan@redhat.com> fix the ci Signed-off-by: myan <myan@redhat.com> ci fix Signed-off-by: Meng Yan <myan@redhat.com> add confluent test in github action Signed-off-by: Meng Yan <myan@redhat.com> add the mis-used test case Signed-off-by: Meng Yan <myan@redhat.com> remove the invalidated bootstrapserver Signed-off-by: Meng Yan <myan@redhat.com> log kafka error message Signed-off-by: myan <myan@redhat.com> review Signed-off-by: myan <myan@redhat.com> Update protocol/kafka_confluent/v2/option.go Co-authored-by: Michael Gasch <15986659+embano1@users.noreply.github.com> Signed-off-by: myan <myan@redhat.com> add the auto recover option Signed-off-by: myan <myan@redhat.com> kafka error handler Signed-off-by: myan <myan@redhat.com> remove the delievery chan Signed-off-by: myan <myan@redhat.com> Update protocol/kafka_confluent/v2/protocol.go Co-authored-by: Michael Gasch <15986659+embano1@users.noreply.github.com> Update protocol/kafka_confluent/v2/protocol.go Co-authored-by: Michael Gasch <15986659+embano1@users.noreply.github.com> Update protocol/kafka_confluent/v2/option.go Co-authored-by: Michael Gasch <15986659+embano1@users.noreply.github.com> Update protocol/kafka_confluent/v2/protocol.go Co-authored-by: Michael Gasch <15986659+embano1@users.noreply.github.com> reply review Signed-off-by: myan <myan@redhat.com> modify the git action Signed-off-by: Meng Yan <myan@redhat.com> reply review Signed-off-by: Meng Yan <myan@redhat.com> reply review Signed-off-by: Meng Yan <myan@redhat.com> handle race condition between sender and closer Signed-off-by: Meng Yan <myan@redhat.com> reply review Signed-off-by: Meng Yan <myan@redhat.com> reply review1 Signed-off-by: Meng Yan <myan@redhat.com> add defer close Signed-off-by: Meng Yan <myan@redhat.com> add comment Signed-off-by: Meng Yan <myan@redhat.com>

embano1

LGTM

THX @yanmxa for getting this over the line!

cc/ @duglin for final overall check

embano1 · 2024-03-29T09:21:19Z

@duglin can you please take a final look just overall whether we missed a README or general requirements for a new protocol binding?

duglin · 2024-03-29T11:28:21Z

Sorry, been swamped. I'll try to give it a look today

duglin · 2024-03-29T11:28:36Z

@lionelvillard too :-)

duglin · 2024-03-29T11:31:09Z

.github/workflows/integration.yaml

@@ -27,6 +27,15 @@ jobs:
          - 9091:9091
          - 9092:9092

+      kafka_confluent:


Should we rename the one above from "kafka" to "kafka_sarama" to be consistent with the rest of the naming?

duglin · 2024-03-29T11:58:22Z

LGTM

I'll let @embano1 hit the button.

yanmxa force-pushed the br_kafka_confluent branch 4 times, most recently from dd224b3 to 5937f4b Compare December 11, 2023 17:28

yanmxa mentioned this pull request Dec 12, 2023

Remove the redundant field from transport message stolostron/multicluster-global-hub#736

Merged

yanmxa force-pushed the br_kafka_confluent branch 2 times, most recently from 2a2bdfd to d196693 Compare December 20, 2023 07:50

yanmxa mentioned this pull request Dec 27, 2023

Replay the unprocessed event from transport when the manager restarts stolostron/multicluster-global-hub#751

Merged

embano1 reviewed Jan 6, 2024

View reviewed changes

test/integration/kafka_confluent/kafka_test.go Outdated Show resolved Hide resolved

embano1 reviewed Jan 6, 2024

View reviewed changes

v2/context/context.go Outdated Show resolved Hide resolved

yanmxa force-pushed the br_kafka_confluent branch from 620866c to 021e19a Compare January 15, 2024 06:55

yanmxa requested a review from embano1 January 23, 2024 06:49

yanmxa mentioned this pull request Jan 26, 2024

Propose the confluent-kafka-go binding for Kafka #1008

Merged

yanmxa force-pushed the br_kafka_confluent branch from a6ee37f to 345f8fc Compare January 30, 2024 07:00

duglin force-pushed the br_kafka_confluent branch from 345f8fc to 3cbec58 Compare January 30, 2024 14:37

duglin force-pushed the br_kafka_confluent branch from e133055 to 8a34203 Compare February 12, 2024 14:28

yanmxa force-pushed the br_kafka_confluent branch from 8a34203 to f2a38c7 Compare February 26, 2024 01:39

embano1 reviewed Mar 10, 2024

View reviewed changes

yanmxa force-pushed the br_kafka_confluent branch from f2a38c7 to 91c54be Compare March 11, 2024 06:35

yanmxa requested a review from embano1 March 11, 2024 13:27

yanmxa force-pushed the br_kafka_confluent branch from 06adc56 to af8817f Compare March 12, 2024 07:18

embano1 reviewed Mar 12, 2024

View reviewed changes

yanmxa force-pushed the br_kafka_confluent branch from e9fafcb to 6836a1c Compare March 25, 2024 13:26

embano1 reviewed Mar 25, 2024

View reviewed changes

protocol/kafka_confluent/v2/protocol.go Outdated Show resolved Hide resolved

embano1 requested changes Mar 25, 2024

View reviewed changes

yanmxa force-pushed the br_kafka_confluent branch 3 times, most recently from 0ac67a3 to 20a4a89 Compare March 26, 2024 02:53

yanmxa force-pushed the br_kafka_confluent branch from 20a4a89 to c70613e Compare March 26, 2024 03:22

yanmxa force-pushed the br_kafka_confluent branch from c70613e to 194cce9 Compare March 26, 2024 03:31

yanmxa force-pushed the br_kafka_confluent branch from 194cce9 to 5cd87ea Compare March 26, 2024 03:33

yanmxa requested a review from embano1 March 26, 2024 03:41

embano1 approved these changes Mar 26, 2024

View reviewed changes

embano1 mentioned this pull request Mar 26, 2024

Panic on Kafka IP change #1028

Open

duglin reviewed Mar 29, 2024

View reviewed changes

embano1 merged commit e6a74ef into cloudevents:main Mar 29, 2024
9 checks passed

yanmxa mentioned this pull request Mar 31, 2024

How to submit message offset asynchronously based on kafka protocol #846

Closed

bread-amammay mentioned this pull request Aug 14, 2024

[kafka_confluent] should propagate partitionkey on an event level #1086

Open

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Confluent kafka binding #988

Confluent kafka binding #988

yanmxa commented Dec 7, 2023 •

edited

Loading

yanmxa commented Dec 12, 2023 •

edited

Loading

yanmxa commented Jan 8, 2024

embano1 commented Jan 23, 2024

lionelvillard commented Jan 24, 2024

yanmxa commented Jan 26, 2024 •

edited

Loading

yanmxa commented Jan 30, 2024

yanmxa commented Mar 6, 2024 •

edited

Loading

embano1 left a comment

embano1 Mar 10, 2024

yanmxa Mar 11, 2024

embano1 Mar 12, 2024

duglin Mar 29, 2024

yanmxa commented Mar 11, 2024 •

edited

Loading

embano1 commented Mar 25, 2024 •

edited

Loading

yanmxa commented Mar 25, 2024

embano1 commented Mar 25, 2024

embano1 commented Mar 25, 2024

yanmxa commented Mar 25, 2024 •

edited

Loading

embano1 left a comment

embano1 commented Mar 25, 2024

embano1 commented Mar 25, 2024

yanmxa commented Mar 26, 2024

yanmxa commented Mar 26, 2024

embano1 left a comment

embano1 commented Mar 29, 2024

duglin commented Mar 29, 2024

duglin commented Mar 29, 2024

duglin Mar 29, 2024

duglin commented Mar 29, 2024

Confluent kafka binding #988

Confluent kafka binding #988

Conversation

yanmxa commented Dec 7, 2023 • edited Loading

yanmxa commented Dec 12, 2023 • edited Loading

yanmxa commented Jan 8, 2024

embano1 commented Jan 23, 2024

lionelvillard commented Jan 24, 2024

yanmxa commented Jan 26, 2024 • edited Loading

yanmxa commented Jan 30, 2024

yanmxa commented Mar 6, 2024 • edited Loading

embano1 left a comment

Choose a reason for hiding this comment

embano1 Mar 10, 2024

Choose a reason for hiding this comment

yanmxa Mar 11, 2024

Choose a reason for hiding this comment

embano1 Mar 12, 2024

Choose a reason for hiding this comment

duglin Mar 29, 2024

Choose a reason for hiding this comment

yanmxa commented Mar 11, 2024 • edited Loading

embano1 commented Mar 25, 2024 • edited Loading

yanmxa commented Mar 25, 2024

embano1 commented Mar 25, 2024

embano1 commented Mar 25, 2024

yanmxa commented Mar 25, 2024 • edited Loading

embano1 left a comment

Choose a reason for hiding this comment

embano1 commented Mar 25, 2024

embano1 commented Mar 25, 2024

yanmxa commented Mar 26, 2024

yanmxa commented Mar 26, 2024

embano1 left a comment

Choose a reason for hiding this comment

embano1 commented Mar 29, 2024

duglin commented Mar 29, 2024

duglin commented Mar 29, 2024

duglin Mar 29, 2024

Choose a reason for hiding this comment

duglin commented Mar 29, 2024

yanmxa commented Dec 7, 2023 •

edited

Loading

yanmxa commented Dec 12, 2023 •

edited

Loading

yanmxa commented Jan 26, 2024 •

edited

Loading

yanmxa commented Mar 6, 2024 •

edited

Loading

yanmxa commented Mar 11, 2024 •

edited

Loading

embano1 commented Mar 25, 2024 •

edited

Loading

yanmxa commented Mar 25, 2024 •

edited

Loading