You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2020/10/21 10:03:24 UTC
[GitHub] [pulsar] wolfstudy opened a new pull request #8327: Add newOuputMessage interface for Go Function
wolfstudy opened a new pull request #8327:
URL: https://github.com/apache/pulsar/pull/8327
Signed-off-by: xiaolong.ran <rx...@apache.org>
Fixes #8266
### Motivation
Implement the context object to be able to send to different topics. Like with the java SDK that allows you to return a future, which makes sending to arbitrary topics easy and fast, we may want to think of an analogous approach
### Modifications
- Add `NewOuputMessage()` interface for Go Function
- Add examples for `NewOuputMessage()`
- Add docs for `NewOuputMessage()`
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] flowchartsman commented on a change in pull request #8327: [Go Functions] Add newOuputMessage interface for Go Function
Posted by GitBox <gi...@apache.org>.
flowchartsman commented on a change in pull request #8327:
URL: https://github.com/apache/pulsar/pull/8327#discussion_r511014789
##########
File path: pulsar-function-go/pf/context.go
##########
@@ -119,6 +120,12 @@ func (c *FunctionContext) GetUserConfMap() map[string]interface{} {
return c.userConfigs
}
+// NewOutputMessage send message to the topic
+// @param topicName: The name of the topic for output message
+func (c *FunctionContext) NewOutputMessage(topicName string) pulsar.Producer {
Review comment:
I can understand the desire for naming parity, but I wonder whether or not it is better to call this something different, given how different the concepts are. In Go, if I see `New<Foo>` I expect that I am going to be receiving a pointer to a new struct of that type, or at least an interface of some kind with that name. In this case, I would expect to get a new Message.
I think I would probably recommend something like the following:
```go
func (c *FunctionContext) SendTopic(topicName string, message *pulsar.ProducerMessage)
```
or even
```go
func (c *FunctionContext) SendTopicMessage(topicName string, message *pulsar.ProducerMessage)
func (c *FunctionContext) SendTopic(topicName string, payload []byte)
```
which, given the way producers are currently packaged, might benefit from a registry of producers per topic:
```go
var producerRegistry = map[string]pulsar.Producer{}
func (gi *goInstance) getProducer(topicName string) (pulsar.Producer, error) {
if found := producerRegistry[topicName]; found != nil {
return found, nil
}
properties := getProperties(getDefaultSubscriptionName(
gi.context.instanceConf.funcDetails.Tenant,
gi.context.instanceConf.funcDetails.Namespace,
gi.context.instanceConf.funcDetails.Name), gi.context.instanceConf.instanceID)
producer, err := gi.client.CreateProducer(pulsar.ProducerOptions{
Topic: topicName,
Properties: properties,
CompressionType: pulsar.LZ4,
BatchingMaxPublishDelay: time.Millisecond * 10,
// Set send timeout to be infinity to prevent potential deadlock with consumer
// that might happen when consumer is blocked due to unacked messages
})
if err != nil {
gi.stats.incrTotalSysExceptions(err)
log.Errorf("create producer error:%s", err.Error())
return nil, err
}
producerRegistry[topicName] = producer
return producer, nil
}
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] sijie commented on pull request #8327: [Go Functions] Add newOuputMessage interface for Go Function
Posted by GitBox <gi...@apache.org>.
sijie commented on pull request #8327:
URL: https://github.com/apache/pulsar/pull/8327#issuecomment-714122933
@nlu90 Can you review this as well?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] nlu90 commented on a change in pull request #8327: [Go Functions] Add newOuputMessage interface for Go Function
Posted by GitBox <gi...@apache.org>.
nlu90 commented on a change in pull request #8327:
URL: https://github.com/apache/pulsar/pull/8327#discussion_r510552758
##########
File path: pulsar-function-go/pf/context.go
##########
@@ -119,6 +120,12 @@ func (c *FunctionContext) GetUserConfMap() map[string]interface{} {
return c.userConfigs
}
+// NewOutputMessage send message to the topic
+// @param topicName: The name of the topic for output message
+func (c *FunctionContext) NewOutputMessage(topicName string) pulsar.Producer {
Review comment:
why a `pulsar.Producer` is returned for this method?
in the corresponding java side, `public <O> TypedMessageBuilder<O> newOutputMessage(String topicName, Schema<O> schema)`
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] wolfstudy commented on a change in pull request #8327: [Go Functions] Add newOuputMessage interface for Go Function
Posted by GitBox <gi...@apache.org>.
wolfstudy commented on a change in pull request #8327:
URL: https://github.com/apache/pulsar/pull/8327#discussion_r511289613
##########
File path: pulsar-function-go/pf/context.go
##########
@@ -119,6 +120,12 @@ func (c *FunctionContext) GetUserConfMap() map[string]interface{} {
return c.userConfigs
}
+// NewOutputMessage send message to the topic
+// @param topicName: The name of the topic for output message
+func (c *FunctionContext) NewOutputMessage(topicName string) pulsar.Producer {
Review comment:
Thanks @flowchartsman feedback. In fact, here we hope to expose the *Producer* interface rather than the `producerMessage` struct. In this way, users can use this interface to maximize their own behavior, including setting `producerOptions` when creating a producer.
Again, `NewOutputMessage()` most fundamental purpose is to intercept the results processed by the Pulsar Function, and then transfer the processing power of this result to the user for customization.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] flowchartsman commented on a change in pull request #8327: [Go Functions] Add newOuputMessage interface for Go Function
Posted by GitBox <gi...@apache.org>.
flowchartsman commented on a change in pull request #8327:
URL: https://github.com/apache/pulsar/pull/8327#discussion_r511020456
##########
File path: pulsar-function-go/pf/context.go
##########
@@ -119,6 +120,12 @@ func (c *FunctionContext) GetUserConfMap() map[string]interface{} {
return c.userConfigs
}
+// NewOutputMessage send message to the topic
+// @param topicName: The name of the topic for output message
+func (c *FunctionContext) NewOutputMessage(topicName string) pulsar.Producer {
Review comment:
Admittedly, there are certainly more idiomatic ways to handle some of these concepts. I am generally against using global variables, especially maps with no synchronization, but from what I can tell from the code all access to this map would be synchronous, so no problems there, and more profound changes to the Go functions SDK would require a PIP, which I'm certainly open to submitting down the road.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] wolfstudy commented on a change in pull request #8327: [Go Functions] Add newOuputMessage interface for Go Function
Posted by GitBox <gi...@apache.org>.
wolfstudy commented on a change in pull request #8327:
URL: https://github.com/apache/pulsar/pull/8327#discussion_r510554485
##########
File path: pulsar-function-go/pf/context.go
##########
@@ -119,6 +120,12 @@ func (c *FunctionContext) GetUserConfMap() map[string]interface{} {
return c.userConfigs
}
+// NewOutputMessage send message to the topic
+// @param topicName: The name of the topic for output message
+func (c *FunctionContext) NewOutputMessage(topicName string) pulsar.Producer {
Review comment:
This is because Java client and Go client have different packaging for producer and message. In the Go client, they are independent structs; in the Java client, they are unified encapsulated into TypedMessageBuilder objects.
But in fact, newOuputMessage essentially intercepts the link that sends messages to `outputTopic` during pulsar function processing. So we need to expose the producer to allow users to send messages out.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] jiazhai commented on pull request #8327: [Go Functions] Add newOuputMessage interface for Go Function
Posted by GitBox <gi...@apache.org>.
jiazhai commented on pull request #8327:
URL: https://github.com/apache/pulsar/pull/8327#issuecomment-714198514
@wolfstudy please help add test for this feature
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] flowchartsman commented on a change in pull request #8327: [Go Functions] Add newOuputMessage interface for Go Function
Posted by GitBox <gi...@apache.org>.
flowchartsman commented on a change in pull request #8327:
URL: https://github.com/apache/pulsar/pull/8327#discussion_r511014789
##########
File path: pulsar-function-go/pf/context.go
##########
@@ -119,6 +120,12 @@ func (c *FunctionContext) GetUserConfMap() map[string]interface{} {
return c.userConfigs
}
+// NewOutputMessage send message to the topic
+// @param topicName: The name of the topic for output message
+func (c *FunctionContext) NewOutputMessage(topicName string) pulsar.Producer {
Review comment:
I can understand the desire for naming parity, but I wonder whether or not it is better to call this something different, given how different the concepts are. In Go, if I see `New<Foo>` I expect that I am going to be receiving a pointer to a new struct of that type, or at least an interface of some kind with that name. In this case, I would expect to get a new Message.
I think I would probably recommend something like the following:
```go
func (c *FunctionContext) SendTopic(topicName string, message *pulsar.ProducerMessage)
```
or even
```go
func (c *FunctionContext) SendTopicMessage(topicName string, message *pulsar.ProducerMessage)
func (c *FunctionContext) SendTopic(topicName string, payload []byte)
```
which, given the way producers are currently packaged, might benefit from a registry of producers per topic:
```go
var producerRegistry = map[string]pulsar.Producer{}
func (gi *goInstance) getProducer(topicName string) (pulsar.Producer, error) {
if found := producerRegistry[topicName]; found != nil {
return found, nil
}
properties := getProperties(getDefaultSubscriptionName(
gi.context.instanceConf.funcDetails.Tenant,
gi.context.instanceConf.funcDetails.Namespace,
gi.context.instanceConf.funcDetails.Name), gi.context.instanceConf.instanceID)
producer, err := gi.client.CreateProducer(pulsar.ProducerOptions{
Topic: topicName,
Properties: properties,
CompressionType: pulsar.LZ4,
BatchingMaxPublishDelay: time.Millisecond * 10,
// Set send timeout to be infinity to prevent potential deadlock with consumer
// that might happen when consumer is blocked due to unacked messages
})
if err != nil {
gi.stats.incrTotalSysExceptions(err)
log.Errorf("create producer error:%s", err.Error())
return nil, err
}
producerRegistry[topicName] = producer
return producer, err
}
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] wolfstudy merged pull request #8327: [Go Functions] Add newOuputMessage interface for Go Function
Posted by GitBox <gi...@apache.org>.
wolfstudy merged pull request #8327:
URL: https://github.com/apache/pulsar/pull/8327
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] wolfstudy commented on pull request #8327: [Go Functions] Add newOuputMessage interface for Go Function
Posted by GitBox <gi...@apache.org>.
wolfstudy commented on pull request #8327:
URL: https://github.com/apache/pulsar/pull/8327#issuecomment-714874664
/pulsarbot run-failure-checks
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org