Initial checkin of PublishMessage/Publish by flowchartsman · Pull Request #9515 · apache/pulsar

flowchartsman · 2021-02-07T06:41:27Z

fixes #9512

Motivation

#9512

Modifications

similar to #9512, but errors are fatal, as with normal message handling. (FEEDBACK REQUESTED))

Verifying this change

None of the tests in the Go code exercise this change, and there were no tests for NewOutputMessage either, except for very basic assertions on a producer.

This change added tests and can be verified as follows:

I will need help testing this, because currently there are no integration tests for go pulsar functions in the SDK code. I am still unfamiliar with how they are integration / unit tested at all outside of the module

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (no)
The public API: (no)
The schema: (no)
The default values of configurations: (no)
The wire protocol: (no)
The rest endpoints: (no)
The admin cli options: (no)
Anything that affects deployment: (no)

Documentation

Does this pull request introduce a new feature? (yes)
If yes, how is the feature documented? (not documented) (draft)

fixes apache#9512

flowchartsman · 2021-02-07T06:41:48Z

feedback CC @freeznet @sijie

flowchartsman · 2021-02-07T06:58:57Z

This is one possible way this could be implemented, sticking to the standards of Go Pulsar functions by sending the message async and not returning error/messageID to the user. The other possible route would be like my original suggestion, and then all sends to topics that are not the output topic would be synchronous. FEEDBACK REQUESTED

sijie · 2021-02-08T05:05:21Z

@freeznet @nlu90 Can you review this?

nlu90 · 2021-02-08T19:56:28Z

pulsar-function-go/pf/instance.go

+		producer.SendAsync(ctx, &message, func(_ pulsar.MessageID, _ *pulsar.ProducerMessage, err error) {
+			if err != nil {
+				goInstance.stats.incrTotalSysExceptions(err)
+				log.Fatal(err)


Do we want to exit the process here or instead return an error to allow users to handle delivering failures?

nlu90 · 2021-02-08T19:58:10Z

pulsar-function-go/examples/publishFunc/publishFunc.go

-
-	producer := fctx.NewOutputMessage(publishTopic)
-	msgID, err := producer.Send(ctx, &pulsar.ProducerMessage{
+	log.Printf("publishing to additional topic %s", publishTopic)


is this for debugging purpose and will be removed when checked in?

This is just the example, so it's only there to show that it's doing something. I wasn't sure what to put here. Recommendations?

zzzming · 2021-02-08T21:28:10Z

pulsar-function-go/pf/context.go

+	Publish        func(topic string, payload []byte)
+	PublishMessage func(topic string, message pulsar.ProducerMessage)


Can you give little more control to the caller with a context and return an error?

Something like func(ctx Context, topic string, payload []byte), since you are creating and passing a context to the lower level anyway.

Return an error instead of log.Fatal crash the process.

@zzzming +1

I originally wrote it that way, but returning an error is not how the rest of the framework works. If there's an error, it dies, relying on metrics/backpressure to alert and remediate. Why should this be any different for multiple output topics? If you look at the conceptual documents for multi-topic dispatch (with python examples), they are not treated any differently as a unit. To me, this suggests they should be treated (and automatically tracked with metrics) the same way, otherwise the user should just create a regular consumer/producer with the regular golang SDK, not a pulsar function.

freeznet · 2021-02-09T07:44:40Z

pulsar-function-go/pf/context.go

-// topic for output message
+// topic for output message (DEPRECATED)
 func (c *FunctionContext) NewOutputMessage(topicName string) pulsar.Producer {
+	log.Warn("NewOutputMessage is deprecated, please use Publish() or PublishMsg()")


We should add a new function like NewProducer(topicName string) (pulsar.Producer, error) or with any proper name. the reason is we need to remain this function exists, and let user know with the log.Warn if they still want this.

Normally I would agree, but as I've discussed elsewhere, the pulsar function SDK API is relatively lightweight by design. The more knobs you give the user to tweak at runtime, the further you get from the control of the process the SDK appears to be designed for. At that point, it is not much different from a regular consumer/producer with deployment, some configuration and some logging taken care of.

freeznet · 2021-02-09T07:45:36Z

pulsar-function-go/pf/context.go

+	Publish        func(topic string, payload []byte)
+	PublishMessage func(topic string, message pulsar.ProducerMessage)


@zzzming +1

codelipenghui · 2022-03-04T07:05:56Z

The pr had no activity for 30 days, mark with Stale label.

github-actions · 2022-03-04T07:06:58Z

@flowchartsman:Thanks for your contribution. For this PR, do we need to update docs?
(The PR template contains info about doc, which helps others know more about the changes. Can you provide doc-related info in this and future PR descriptions? Thanks)

tisonkun · 2022-11-11T09:04:04Z

Closed as the development of the Go client has been permanently moved to https://github.com/apache/pulsar-client-go.

tisonkun · 2022-11-11T09:05:22Z

Oops. It's about the Go Functions, not the Go client.

But still, there're several conflicts so please feel free to resubmit the patch.

Initial checkin of PublishMessage/Publish

e30e260

fixes apache#9512

sijie assigned flowchartsman Feb 8, 2021

sijie added this to the 2.8.0 milestone Feb 8, 2021

sijie added area/function component/go labels Feb 8, 2021

nlu90 reviewed Feb 8, 2021

View reviewed changes

zzzming reviewed Feb 8, 2021

View reviewed changes

freeznet requested changes Feb 9, 2021

View reviewed changes

flowchartsman mentioned this pull request Feb 19, 2021

[Go Functions] Cache producers with sync.Map #9513

Closed

codelipenghui removed this from the 2.8.0 milestone May 21, 2021

codelipenghui added the lifecycle/stale label Mar 4, 2022

github-actions bot added the doc-label-missing label Mar 4, 2022

tisonkun closed this Nov 11, 2022

		Publish func(topic string, payload []byte)
		PublishMessage func(topic string, message pulsar.ProducerMessage)

Conversation

flowchartsman commented Feb 7, 2021

Motivation

Modifications

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Uh oh!

flowchartsman commented Feb 7, 2021

Uh oh!

flowchartsman commented Feb 7, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

sijie commented Feb 8, 2021

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

flowchartsman Feb 18, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

codelipenghui commented Mar 4, 2022

Uh oh!

github-actions bot commented Mar 4, 2022

Uh oh!

tisonkun commented Nov 11, 2022

Uh oh!

tisonkun commented Nov 11, 2022

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

7 participants

flowchartsman commented Feb 7, 2021 •

edited

Loading

flowchartsman Feb 18, 2021 •

edited

Loading