PIP-15: Pulsar Functions #1314

sijie · 2018-03-01T08:29:21Z

This is the PR for PIP-15: https://github.com/apache/incubator-pulsar/wiki/PIP-15:-Pulsar-Functions

The branch for review: https://github.com/sijie/incubator-pulsar/tree/pulsar_functions

All code changes are made in a separated module under directory pulsar-functions : https://github.com/sijie/incubator-pulsar/tree/pulsar_functions/pulsar-functions

Authors: @srkukarni @jerrypeng @sijie

sijie · 2018-03-01T08:33:47Z

it is probably hard for review. we have tried to make the changes happen under pulsar-functions only. the only two places changing outside of pulsar-functions:

adding a module pulsar-functions in pom.xml
fixed a typo issue in pulsar-client-tools-shaded (a separated change is also sent for review Fix the typo in shaded client-tools module #1313, will rebase after Fix the typo in shaded client-tools module #1313 is merged)

ivankelly

Impossible to do thorough review for 25K+ LoC, but it only adds 60s to test times, and it looks good from a quick scan, so I say +1.

(full disclosure: I work for streamlio)

merlimat

👍
Just left few comments that can be addressed later.

merlimat · 2018-03-02T22:21:30Z

pulsar-functions/dist/pom.xml

@@ -0,0 +1,109 @@
+<!--


Is this pom still needed or should we just rollover into parent dist module?

yes. we should rollover into the parent dist module. probably better to do after merging this, to avoid touch files outside this functions module.

merlimat · 2018-03-02T22:23:05Z

pulsar-functions/instance/pom.xml

+  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+
+  <properties>


I'd prefer to set all package version (and variables) in top-level pom so that we have 1 single place to check/update, unless there's a specific requirement.

yes. will do after merging

merlimat · 2018-03-02T22:24:22Z

pulsar-functions/instance/pom.xml

+      <version>3.5.1</version>
+    </dependency>
+
+    <dependency>


Same for this versions, they should go to top-level pom. For protobuf, we could have 2 variables protobuf2.version and protobuf3.version.

yes. will do after mering.

merlimat · 2018-03-02T22:25:52Z

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

@@ -0,0 +1,111 @@
+#!/usr/bin/env python
+#
+# Licensed to the Apache Software Foundation (ASF) under one


ASF headers looks repeated twice in some Python files

nice catch. will try to fix it or do it in a separate PR

merlimat · 2018-03-02T22:29:15Z

...ns/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/CounterFunction.java

+
+public class CounterFunction implements PulsarFunction<String, Void> {
+    @Override
+    public Void process(String input, Context context) throws Exception {


Would it make sense to add a specialization of PulsarFunction interface for a void function?

In that way, we won't have the Void return type with the meaningless return null .

That is a good question. I think it falls into a question, do we support java.util.Consumer, which is a void native function.

/cc @srkukarni

Yes java.util.Consumer is fine and probably easy to support (without SDK), though it would need a similar void correspective of PulsarFunction, like PulsarConsumer... which might get confusing.. :)

merlimat · 2018-03-02T22:34:19Z

pulsar-functions/python-examples/exclamation.py

+
+from pulsarfunction import pulsar_function
+
+class Exclamation(pulsar_function.PulsarFunction):


I think the package here looks a bit redundant.

This could be shortened into:

from pulsar import Function class Exclamation(Function): def process(self, input, context): return input + '!'

or

impor pulsar class Exclamation(pulsar.Function): def process(self, input, context): return input + '!'

We just need to figure out if there will be a clash with the pulsar-client Python library.

/cc @srkukarni

merlimat · 2018-03-02T22:37:09Z

.../src/test/java/org/apache/pulsar/functions/worker/rest/api/v1/FunctionApiV1ResourceTest.java

+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.functions.worker.rest.api.v1;


Since we're already moving REST api to v2 (and changing major version or project to v2), I think we should consider to use v2 for functions as well, to avoid confusion.

a good point. will fix it.

merlimat · 2018-03-03T19:10:52Z

Ok, since there were no objection, we're going to merge this PR.

* Create pulsar-functions module (#1) * Create pulsar-functions module * rename `sdk` package to `api` * Added the first cut of the Java interface for Pulsar functions (#2) * Fix void type handling

* Create pulsar-functions module (#1) * Create pulsar-functions module * rename `sdk` package to `api` * Added the first cut of the Java interface for Pulsar functions (#2) * Added a way to limit the number of outstanding tuples read but still not executed

* Relocate instance and spawner to be under `runtime` package * fix the pom files and travis build

* Create pulsar-functions module (#1) * Create pulsar-functions module * rename `sdk` package to `api` * Added the first cut of the Java interface for Pulsar functions (#2) * Introduced the concept of systemException to differentiate from userException

* Create pulsar-functions module (#1) * Create pulsar-functions module * rename `sdk` package to `api` * Added the first cut of the Java interface for Pulsar functions (#2) * Refactor cli to establish a base FunctionCommand interface that all subcommands can piggyback

…JavaInstance (apache#30) ### Motivation SerDe is user defined class which used for serializing and deserializing objects for functions. So the initialization of a SerDe class should happen along with Function class and done by the context class loader. Also SerDe should be a generic-typed interface and the type should be consistent with function type. ### Modifications - make SerDe a generic-typed interface - move SerDe initialization to JavaInstance and use the thread context class loader - verify SerDe type - move reflection related code to a util class `Reflections` ### Result - SerDe is a generic-typed interface - JavaInstance initializes both function and serdes and verifies the types.

### Motivation input type can be different output type. so we need two serdes for each function. ### Modifications - add inputSerde and outputSerde - verify both input and output serde to make sure their types are consistent with function types ### Result user run functions with input/output serdes.

* Create pulsar-functions module (#1) * Create pulsar-functions module * rename `sdk` package to `api` * Added the first cut of the Java interface for Pulsar functions (#2) * Made our function hierarchy match pulsar's * Replace cluster with tenant

* Create pulsar-functions module (#1) * Create pulsar-functions module * rename `sdk` package to `api` * Added the first cut of the Java interface for Pulsar functions (#2) * Added the first cut of the pulsar submit * Working

* Create pulsar-functions module (#1) * Create pulsar-functions module * rename `sdk` package to `api` * Added the first cut of the Java interface for Pulsar functions (#2) * Fix compilation issue

* Create pulsar-functions module (#1) * Create pulsar-functions module * rename `sdk` package to `api` * Added the first cut of the Java interface for Pulsar functions (#2) * Made ack async

- the package will be generated under `dist/target/`

…#262) * Support parallelism parameter while running functions locally * Fixed the generation of java and python files

…#264)

* Move the thread construction to start to avoid illegalstate * Add more checks to see if thread var is init

* Use `distributedlog-core-shaded` in pulsar worker * revert to db ledger storage * Include netty-all * Fix serviceUrl for functions cli

- downloader is writing the package to a temp location - after a package is downloaded, a hardlink is created to link to the temp file. hardlink will ensure one createLink succeed. - after the hardlink is created, the temp files can be deleted. After this change, it allows concurrent downloading without interleaving with each other.

…apache#269) Example: ``` (CONNECTING) /> (CONNECTED) /> ls pulsar/functions/test test-namespace (CONNECTED) /> ls pulsar/functions/test/test-namespace function11 function12 function13 (CONNECTED) /> ls pulsar/functions/test/test-namespace/function11 8e4dcc81-3b00-4bf9-a489-1823b5514788-pulsar-functions-api-examples.jar (CONNECTED) /> ls pulsar/functions/test/test-namespace/function12 0ee8e267-74cb-4178-99ea-6c723298223c-pulsar-functions-api-examples.jar (CONNECTED) /> ls pulsar/functions/test/test-namespace/function13 8f1dea11-0070-434b-ba97-a42b44d8063c-pulsar-functions-api-examples.jar (CONNECTED) /> ```

The `compile` scope in dependency plugin means : compile, provided, and system dependencies

apache#273) * Create ProcessBuilder at during start() so that start() can be called multiple times from Spawner * Fix unittest

…ache#275)

- Remove function-composition

…cated license headers This addresses some comments in pulsar functions PR apache#1314

…cated license headers (#1324) This addresses some comments in pulsar functions PR #1314

* Move pulsar functions dependency version to root pom and remove duplicated license headers This addresses some comments in pulsar functions PR #1314 * shade worker * Fix broken master * Upgrade the bookkeeper storage client dependency to the official bookkeeper version This removes the temp dependency in `pulsar-functions-instance` * set `protobuf2.version` in pulsar-common * provide a shaded worker * include worker dependency at broker * Embeded function worker at broker * rename 'function worker' to 'functions worker' * add "--no-functions-worker" for pulsar-client-cpp tests

* Move pulsar functions dependency version to root pom and remove duplicated license headers This addresses some comments in pulsar functions PR #1314 * shade worker * Fix broken master * Upgrade the bookkeeper storage client dependency to the official bookkeeper version This removes the temp dependency in `pulsar-functions-instance` * set `protobuf2.version` in pulsar-common * provide a shaded worker * include worker dependency at broker * Embeded function worker at broker * rename 'function worker' to 'functions worker' * add "--no-functions-worker" for pulsar-client-cpp tests * Integrate function cli into pulsar-admin cli - rename `pulsar-client-tools-shaded` to `pulsar-client-admin-shaded-for-functions`, because this module is used by functions only to avoid protobuf conflicts - move protobuf3 references to Utils, so it won't be referenced out side of pulsar-functions - integrate function cli into pulsar-admin cli * Fix license header issues * Fixed ZK cache test exectutor configuration. Fixes #1338

…1332) * Move pulsar functions dependency version to root pom and remove duplicated license headers This addresses some comments in pulsar functions PR #1314 * shade worker * Fix broken master * Upgrade the bookkeeper storage client dependency to the official bookkeeper version This removes the temp dependency in `pulsar-functions-instance` * set `protobuf2.version` in pulsar-common * provide a shaded worker * include worker dependency at broker * Embeded function worker at broker * rename 'function worker' to 'functions worker' * add "--no-functions-worker" for pulsar-client-cpp tests * Integrate function cli into pulsar-admin cli - rename `pulsar-client-tools-shaded` to `pulsar-client-admin-shaded-for-functions`, because this module is used by functions only to avoid protobuf conflicts - move protobuf3 references to Utils, so it won't be referenced out side of pulsar-functions - integrate function cli into pulsar-admin cli * Merge pulsar-functions dist package into pulsar binary distribution * Fix license header issues * Fixed ZK cache test exectutor configuration. Fixes #1338

sijie added the type/feature The PR added a new feature or issue requested a new feature label Mar 1, 2018

sijie added this to the 2.0.0-incubating milestone Mar 1, 2018

sijie requested review from merlimat, rdhabalia, jai1, joefk, nkurihar, yush1ga and massakam March 1, 2018 08:29

ivankelly approved these changes Mar 1, 2018

View reviewed changes

merlimat added the area/function label Mar 2, 2018

sijie mentioned this pull request Mar 2, 2018

Introduce a pulsar log4j2 appender #1316

Merged

merlimat approved these changes Mar 2, 2018

View reviewed changes

srkukarni and others added 15 commits March 3, 2018 16:44

Fix tests (apache#24)

5f3b93e

* Create pulsar-functions module (#1) * Create pulsar-functions module * rename `sdk` package to `api` * Added the first cut of the Java interface for Pulsar functions (#2) * Fix void type handling

Buffer limit (apache#25)

f883f16

* Create pulsar-functions module (#1) * Create pulsar-functions module * rename `sdk` package to `api` * Added the first cut of the Java interface for Pulsar functions (#2) * Added a way to limit the number of outstanding tuples read but still not executed

Relocate packages (apache#21)

5adc066

* Relocate instance and spawner to be under `runtime` package * fix the pom files and travis build

System exception (apache#26)

117f816

* Create pulsar-functions module (#1) * Create pulsar-functions module * rename `sdk` package to `api` * Added the first cut of the Java interface for Pulsar functions (#2) * Introduced the concept of systemException to differentiate from userException

Use the thread context classloader for loading functions (apache#28)

13bbbf7

relocate SerDe classes to serde package and add tests (apache#29)

316f834

Nomenclature (apache#32)

4dd4089

* Create pulsar-functions module (#1) * Create pulsar-functions module * rename `sdk` package to `api` * Added the first cut of the Java interface for Pulsar functions (#2) * Made our function hierarchy match pulsar's * Replace cluster with tenant

Cleanup (apache#34)

3227065

Fix compilation (apache#35)

da66b2d

* Create pulsar-functions module (#1) * Create pulsar-functions module * rename `sdk` package to `api` * Added the first cut of the Java interface for Pulsar functions (#2) * Fix compilation issue

Make producer and consumer async

3f7d304

Asyncack (apache#37)

7fbe342

* Create pulsar-functions module (#1) * Create pulsar-functions module * rename `sdk` package to `api` * Added the first cut of the Java interface for Pulsar functions (#2) * Made ack async

srkukarni and others added 20 commits March 3, 2018 16:44

Added abiltiy to specify broker on cmdline for localrun (apache#259)

7370fbc

fix bug in scheduler (apache#260)

dc8f39c

Moved localrun to process based runtime (apache#261)

9114ddc

Provide a distribution package for pulsar functions (apache#263)

e5ed2ed

- the package will be generated under `dist/target/`

Support parallelism parameter while running functions locally (apache…

e417c57

…#262) * Support parallelism parameter while running functions locally * Fixed the generation of java and python files

Fix pulsar-functions on locating java & python instance files (apache…

68e1f3c

…#264)

Use distributedlog-core-shaded in pulsar worker (apache#257)

981e472

Move the thread construction to start to avoid illegalstate (apache#266)

6652c5a

* Move the thread construction to start to avoid illegalstate * Add more checks to see if thread var is init

Fix serviceUrl for functions cli (apache#267)

322f3af

* Use `distributedlog-core-shaded` in pulsar worker * revert to db ledger storage * Include netty-all * Fix serviceUrl for functions cli

Resolve rebase conflicts

82bccd1

Build classpath using compile scope (apache#272)

ae1ac31

The `compile` scope in dependency plugin means : compile, provided, and system dependencies

Create ProcessBuilder at during start() so that start() can be called… (

2e49646

apache#273) * Create ProcessBuilder at during start() so that start() can be called multiple times from Spawner * Fix unittest

If no brokerUrl is specified use localhost (apache#274)

6ec05f9

fix scaling up and down as well as updating a function in general (ap…

46bc558

…ache#275)

adding SchedulerManager tests (apache#276)

03df440

Better variable safety (apache#277)

433eb8b

Effectively once not yet supported in Python (apache#278)

5245529

- Revert .travis.yml

12dac7e

- Remove function-composition

sijie force-pushed the pulsar_functions branch from c1e089d to 12dac7e Compare March 4, 2018 00:53

merlimat merged commit 12dac7e into apache:master Mar 4, 2018

sijie added a commit to sijie/pulsar that referenced this pull request Mar 4, 2018

Move pulsar functions dependency version to root pom and remove dupli…

641df06

…cated license headers This addresses some comments in pulsar functions PR apache#1314

This was referenced Mar 4, 2018

Move pulsar functions dependency version to root pom and remove duplicated license headers #1324

Merged

Rename pulsar function api from V1 to V2 #1325

Merged

Refactored PulsarAdmin to use builder for configuration #1327

Merged

sijie added a commit that referenced this pull request Mar 5, 2018

Move pulsar functions dependency version to root pom and remove dupli…

585ea13

…cated license headers (#1324) This addresses some comments in pulsar functions PR #1314

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

PIP-15: Pulsar Functions #1314

PIP-15: Pulsar Functions #1314

sijie commented Mar 1, 2018

sijie commented Mar 1, 2018

ivankelly left a comment

merlimat left a comment

merlimat Mar 2, 2018

sijie Mar 3, 2018

merlimat Mar 2, 2018

sijie Mar 3, 2018

merlimat Mar 2, 2018

sijie Mar 3, 2018

merlimat Mar 2, 2018

sijie Mar 3, 2018

merlimat Mar 2, 2018

sijie Mar 3, 2018

merlimat Mar 3, 2018

merlimat Mar 2, 2018

sijie Mar 3, 2018

merlimat Mar 2, 2018

sijie Mar 3, 2018

merlimat commented Mar 3, 2018


		from pulsarfunction import pulsar_function

		class Exclamation(pulsar_function.PulsarFunction):

PIP-15: Pulsar Functions #1314

PIP-15: Pulsar Functions #1314

Conversation

sijie commented Mar 1, 2018

sijie commented Mar 1, 2018

ivankelly left a comment

Choose a reason for hiding this comment

merlimat left a 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

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

merlimat commented Mar 3, 2018