Set key for message when using function publish #4005

jerrypeng · 2019-04-08T19:41:57Z

Motivation

In java functions, when a user calls publish, the outgoing message does not have a key set even though in the input message has a key. By default, set the same key for the outgoing publish method. This is the behavior when sending messages that are returned by the function

In python functions, the message key is never set. Make python functions consistent with java functions

srkukarni · 2019-04-08T21:26:07Z

pulsar-client-cpp/python/pulsar/functions/context.py

Not sure we want to always do it this way. For example, what if we have multiple input topics with different keys, etc.
Would it be better to explicitly have an extra parameter?(Both for java and py)

I think that should be the default behavior if the user doesn't explicitly specify the key since that is the behavior for objects returned from the function. We can add an additional method in which the user can explicitly specify a key

sijie · 2019-04-09T00:05:45Z

pulsar-client-cpp/python/pulsar/functions/context.py

I am not a python expert. but this PR doesn't seem to specify partition_key in the method signature. how does that work?

@sijie I have updated the PR

can we add a ut or integration test for this? this raises a huge concern to me - we update the implementation but didn't change the interface definition. how can we guarantee the fix is correct?

@sijie you reviewed the PR when I was in the middle of updating the PR due to discussions and comments around this PR. The original PR did not modify any of the interfaces, but after a discussion with @srkukarni and @merlimat, we decided it was best if we let the user choose what the partition key is. That said, I will add some tests to it

actually @sijie if you look at the commit you commented on. The signature of contextimpl.publish didn't change. Thus context.py did not need to be updated

Sure I might be commenting in between commits. but my question is still there : how can we guarantee the fix is correct? where are the tests for this change?

jerrypeng · 2019-04-09T03:17:22Z

@sijie in my previous comment, I already said I was going to add the tests. Please read my responses in their entirety before responding.

…

On Mon, Apr 8, 2019 at 8:07 PM Sijie Guo ***@***.***> wrote: ***@***.**** commented on this pull request. ------------------------------ In pulsar-client-cpp/python/pulsar/functions/context.py <#4005 (comment)>: > @@ -126,7 +126,8 @@ def record_metric(self, metric_name, metric_value): @AbstractMethod def publish(self, topic_name, message, serde_class_name="serde.IdentitySerDe", properties=None, compression_type=None, callback=None): Sure I might be commenting in between commits. but my question is still there : how can we guarantee the fix is correct? where are the tests for this change? — You are receiving this because you were assigned. Reply to this email directly, view it on GitHub <#4005 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/ADcir6hlv1dcxwEj9LyuwhTw3xsfOlI-ks5vfAPtgaJpZM4ci9Wj> .

sijie · 2019-04-09T03:18:18Z

@jerrypeng got it now. thank you

…t_publish_set_key

…ext_publish_set_key

jerrypeng · 2019-04-11T21:34:52Z

@merlimat @jerrypeng @sijie please review this PR again. I have rewritten this PR to use the change @merlimat made for MessageBuilder. I added also included several tests

merlimat

👍

srkukarni

do you want to add an integration test to cover this feature?

sijie

@jerrypeng overall looks good to me. left a few comments there. one of the main questions is around the partition_key you added to the output message. I am not sure if that's correct, unless I am missing some context there.

sijie · 2019-04-12T23:54:26Z

pulsar-client-cpp/python/pulsar/functions/context.py

+    pass
+
+  @abstractmethod
+  def publish(self, topic_name, message, serde_class_name="serde.IdentitySerDe", compression_type=None, callback=None, message_conf=None):


nit: any reason why do we have two methods here? doesn't the named argument message_conf=None is good enough?

@sijie are you talking about

def publish(self, topic_name, message, serde_class_name="serde.IdentitySerDe", properties=None, compression_type=None, callback=None):

and

def publish(self, topic_name, message, serde_class_name="serde.IdentitySerDe", compression_type=None, callback=None, message_conf=None):

The former is for backwards compatibility purposes

I think in python you don't need two methods for backwards compatibility though. one method with named argument is good enough

sijie · 2019-04-12T23:58:23Z

pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Context.java

+     *
+     * @return A future that completes when the framework is done publishing the message
+     */
+    <O> CompletableFuture<Void> publish(String topicName, O object, String schemaOrSerdeClassName, Map<String, Object> messageConf);


isn't it better to expose publish(String topicName, O object, Schema<T> schema, Map<String, Object> messageConf)?

I think we do some caching with schemaOrSerdeClassName so that we don't instantiate the schema/SerDe every single type. And we use schemaOrSerdeClassName as the key for the cache. Not greatest user experience but that is what is happening now

sijie · 2019-04-13T00:02:27Z

pulsar-functions/instance/src/main/python/python_instance.py

      if output_bytes is not None:
        props = {"__pfn_input_topic__" : str(msg.topic), "__pfn_input_msg_id__" : base64ify(msg.message.message_id().serialize())}
-        self.producer.send_async(output_bytes, partial(self.done_producing, msg.consumer, msg.message, self.producer.topic()), properties=props)
+        self.producer.send_async(output_bytes, partial(self.done_producing, msg.consumer, msg.message, self.producer.topic()), properties=props, partition_key=msg.message.partition_key())


we are using the partition_key from the input message as the partition_key for the output message. it doesn't seem to be correct to me.

any reason why you are setting this key here?

To be consistent with Java, but I will remove this for now

sijie · 2019-04-13T00:05:34Z

...s/src/main/java/org/apache/pulsar/functions/api/examples/PublishFunctionWithMessageConf.java

+        if (context.getCurrentRecord().getKey().isPresent()) {
+            messageConf.put("key", context.getCurrentRecord().getKey().get());
+        }
+        messageConf.put("eventTime", System.currentTimeMillis());


nit: can we use the constants defined in TypedMessageBuilder.java

will change

jerrypeng · 2019-04-13T01:30:51Z

@sijie thanks for the review. I have updated the PR.

jerrypeng · 2019-04-13T03:05:34Z

rerun java8 tests
rerun integration tests

jerrypeng · 2019-04-13T03:52:27Z

rerun java8 tests
rerun integration tests

jerrypeng · 2019-04-13T03:52:41Z

rerun java8 tests

jerrypeng · 2019-04-13T03:53:18Z

rerun integration tests

jerrypeng · 2019-04-13T04:55:25Z

rerun java8 tests

### Motivation Pulsar release 2.4.0 supports setting partition key for messages when using function publish, as described in this PR: #4005 The code is updated but the doc is not updated accordingly. ### Modifications 1: Update the doc based on the code updated. 2: also add some legacy missing methods in the context class of the python functions according to the code. 3: This update is only for Pulsar 2.4.0 release. No need to update docs in 2.4.1 and later releases as these docs have been updated.

### Motivation Pulsar release 2.4.0 supports setting partition key for messages when using function publish, as described in this PR: apache#4005 The code is updated but the doc is not updated accordingly. ### Modifications 1: Update the doc based on the code updated. 2: also add some legacy missing methods in the context class of the python functions according to the code. 3: This update is only for Pulsar 2.4.0 release. No need to update docs in 2.4.1 and later releases as these docs have been updated.

jerrypeng added the area/function label Apr 8, 2019

jerrypeng added this to the 2.3.2 milestone Apr 8, 2019

jerrypeng self-assigned this Apr 8, 2019

jerrypeng requested review from merlimat, sijie and srkukarni April 8, 2019 19:41

srkukarni reviewed Apr 8, 2019

View reviewed changes

sijie requested changes Apr 9, 2019

View reviewed changes

merlimat and others added 10 commits April 9, 2019 16:19

Allow to configure TypedMessageBuilder through a Map conf object

2e84172

Use constants for message confs

8885b0c

Reverted previous change

b5792b6

Use Long instead of Number

90aa999

Set key for message when using function publish

e59e730

fix unit test

7f0d7c5

fix python test

46ce383

improving impl

993c382

Merge branch 'message-conf' of github.com:merlimat/pulsar into contex…

95d1cfd

…t_publish_set_key

improving implementation

ba9b884

jerrypeng force-pushed the context_publish_set_key branch from 6cedfa5 to ba9b884 Compare April 11, 2019 00:54

jerrypeng added 2 commits April 10, 2019 22:18

Merge branch 'master' of github.com:apache/incubator-pulsar into cont…

dea473a

…ext_publish_set_key

add tests and examples

35a4640

jerrypeng requested a review from sijie April 11, 2019 21:34

jerrypeng requested a review from srkukarni April 11, 2019 21:35

jerrypeng added 2 commits April 11, 2019 14:49

fix bug

45f79de

fix bug

03be813

merlimat approved these changes Apr 12, 2019

View reviewed changes

fixing comments

34361ae

fix example

bcae09c

srkukarni approved these changes Apr 12, 2019

View reviewed changes

sijie requested changes Apr 13, 2019

View reviewed changes

addressing comments

d58330e

sijie approved these changes Apr 13, 2019

View reviewed changes

fix function

332405d

jerrypeng merged commit 23b1418 into apache:master Apr 13, 2019

sijie mentioned this pull request Apr 23, 2019

[issue#4042] improve java functions API #4093

Merged

merlimat modified the milestones: 2.3.2, 2.4.0 May 19, 2019

Huanli-Meng mentioned this pull request Jul 2, 2020

[Doc]--Set partition key for message when using function publish #7431

Merged

Set key for message when using function publish #4005

Set key for message when using function publish #4005

Uh oh!

Conversation

jerrypeng commented Apr 8, 2019

Motivation

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!

jerrypeng Apr 9, 2019 • 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!

jerrypeng Apr 9, 2019 • 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!

jerrypeng commented Apr 9, 2019 via email

Uh oh!

sijie commented Apr 9, 2019

Uh oh!

jerrypeng commented Apr 11, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

merlimat left a comment

Choose a reason for hiding this comment

Uh oh!

srkukarni left a comment

Choose a reason for hiding this comment

Uh oh!

sijie left a comment

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!

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!

jerrypeng commented Apr 13, 2019

Uh oh!

jerrypeng commented Apr 13, 2019

Uh oh!

jerrypeng commented Apr 13, 2019

Uh oh!

jerrypeng commented Apr 13, 2019

Uh oh!

jerrypeng commented Apr 13, 2019

Uh oh!

jerrypeng commented Apr 13, 2019

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

jerrypeng Apr 9, 2019 •

edited

Loading

jerrypeng Apr 9, 2019 •

edited

Loading

jerrypeng commented Apr 11, 2019 •

edited

Loading