[GitHub] sijie closed pull request #2369: If --customSerdeInputs is specified, please update the right variable
sijie closed pull request #2369: If --customSerdeInputs is specified, please
update the right variable
URL: https://github.com/apache/incubator-pulsar/pull/2369
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java
b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java
index b11dabee82..29619adad9 100644
---
a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java
+++
b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java
@@ -316,6 +316,7 @@ private void mergeArgs() {
if (!StringUtils.isBlank(DEPRECATED_topicsPattern)) topicsPattern
= DEPRECATED_topicsPattern;
if (!StringUtils.isBlank(DEPRECATED_logTopic)) logTopic =
DEPRECATED_logTopic;
if (!StringUtils.isBlank(DEPRECATED_outputSerdeClassName))
outputSerdeClassName = DEPRECATED_outputSerdeClassName;
+if (!StringUtils.isBlank(DEPRECATED_customSerdeInputString))
customSerdeInputString = DEPRECATED_customSerdeInputString;
if (!StringUtils.isBlank(DEPRECATED_fnConfigFile)) fnConfigFile =
DEPRECATED_fnConfigFile;
if (DEPRECATED_processingGuarantees !=
FunctionConfig.ProcessingGuarantees.ATLEAST_ONCE) processingGuarantees =
DEPRECATED_processingGuarantees;
if (!StringUtils.isBlank(DEPRECATED_userConfigString))
userConfigString = DEPRECATED_userConfigString;
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[incubator-pulsar] branch master updated: If --customSerdeInputs is specified, please update the right variable (#2369)
This is an automated email from the ASF dual-hosted git repository.
sijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 4810a26 If --customSerdeInputs is specified, please update the right
variable (#2369)
4810a26 is described below
commit 4810a26e370f3a253fe98222d486433a91f04cd1
Author: Sanjeev Kulkarni
AuthorDate: Mon Aug 13 23:58:22 2018 -0700
If --customSerdeInputs is specified, please update the right variable
(#2369)
### Motivation
When we made the switch to get away from camel-case arguments for
functions, we moved the camel case arguments as deprecated, so users can still
use them but they are hidden. But that change missed to take
--customSerdeInputs into account. This fixes that ommision
### Modifications
Describe the modifications you've done.
### Result
After your change, what will change.
---
.../src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java | 1 +
1 file changed, 1 insertion(+)
diff --git
a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java
b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java
index dd1ef3d..5f52313 100644
---
a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java
+++
b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java
@@ -317,6 +317,7 @@ public class CmdFunctions extends CmdBase {
if (!StringUtils.isBlank(DEPRECATED_topicsPattern)) topicsPattern
= DEPRECATED_topicsPattern;
if (!StringUtils.isBlank(DEPRECATED_logTopic)) logTopic =
DEPRECATED_logTopic;
if (!StringUtils.isBlank(DEPRECATED_outputSerdeClassName))
outputSerdeClassName = DEPRECATED_outputSerdeClassName;
+if (!StringUtils.isBlank(DEPRECATED_customSerdeInputString))
customSerdeInputString = DEPRECATED_customSerdeInputString;
if (!StringUtils.isBlank(DEPRECATED_fnConfigFile)) fnConfigFile =
DEPRECATED_fnConfigFile;
if (DEPRECATED_processingGuarantees !=
FunctionConfig.ProcessingGuarantees.ATLEAST_ONCE) processingGuarantees =
DEPRECATED_processingGuarantees;
if (!StringUtils.isBlank(DEPRECATED_userConfigString))
userConfigString = DEPRECATED_userConfigString;
[GitHub] sijie closed pull request #2368: Fix Copy Command in Pulsar IO Getting Started
sijie closed pull request #2368: Fix Copy Command in Pulsar IO Getting Started
URL: https://github.com/apache/incubator-pulsar/pull/2368
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/site2/docs/io-quickstart.md b/site2/docs/io-quickstart.md
index ceefdeb1f2..042ce8e48b 100644
--- a/site2/docs/io-quickstart.md
+++ b/site2/docs/io-quickstart.md
@@ -66,7 +66,7 @@ $ tar xvfz
/path/to/apache-pulsar-io-connectors-{{pulsar:version}}-bin.tar.gz
// you will find a directory named
`apache-pulsar-io-connectors-{{pulsar:version}}` in the pulsar directory
// then copy the connectors
-$ cd apache-pulsar-io-connectors-{{pulsar:version}}/connectors connectors
+$ cp -r apache-pulsar-io-connectors-{{pulsar:version}}/connectors connectors
$ ls connectors
pulsar-io-aerospike-{{pulsar.version}}.nar
diff --git
a/site2/website/versioned_docs/version-2.1.0-incubating/io-quickstart.md
b/site2/website/versioned_docs/version-2.1.0-incubating/io-quickstart.md
index 37868db6e7..069e2bb567 100644
--- a/site2/website/versioned_docs/version-2.1.0-incubating/io-quickstart.md
+++ b/site2/website/versioned_docs/version-2.1.0-incubating/io-quickstart.md
@@ -67,7 +67,7 @@ $ tar xvfz
/path/to/apache-pulsar-io-connectors-{{pulsar:version}}-bin.tar.gz
// you will find a directory named
`apache-pulsar-io-connectors-{{pulsar:version}}` in the pulsar directory
// then copy the connectors
-$ cd apache-pulsar-io-connectors-{{pulsar:version}}/connectors connectors
+$ cp -r apache-pulsar-io-connectors-{{pulsar:version}}/connectors connectors
$ ls connectors
pulsar-io-aerospike-{{pulsar.version}}.nar
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[incubator-pulsar] branch master updated: change 'cd' to 'cp -r' (#2368)
This is an automated email from the ASF dual-hosted git repository.
sijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new d5c7a9b change 'cd' to 'cp -r' (#2368)
d5c7a9b is described below
commit d5c7a9b95fcc21def4f1bd3c3458a965408db35f
Author: Yuto Furuta
AuthorDate: Tue Aug 14 15:56:58 2018 +0900
change 'cd' to 'cp -r' (#2368)
### Motivation
In https://pulsar.incubator.apache.org/docs/en/io-quickstart/ ,
`connectors` directory cannot be copied because copy command is wrong.
> // you will find a directory named
`apache-pulsar-io-connectors-2.1.0-incubating` in the pulsar directory
> // then copy the connectors
>
> $ cd apache-pulsar-io-connectors-2.1.0-incubating/connectors connectors
### Modifications
・Change `cd` to `cp -r`
### Result
`connectors` directory can be copied successfully.
```
$ cp -r apache-pulsar-io-connectors-2.1.0-incubating/connectors connectors
$ ls connectors
pulsar-io-aerospike-2.1.0-incubating.nar
pulsar-io-kinesis-2.1.0-incubating.nar
pulsar-io-cassandra-2.1.0-incubating.nar
pulsar-io-rabbitmq-2.1.0-incubating.nar
pulsar-io-kafka-2.1.0-incubating.nar
pulsar-io-twitter-2.1.0-incubating.nar
```
---
site2/docs/io-quickstart.md| 2 +-
site2/website/versioned_docs/version-2.1.0-incubating/io-quickstart.md | 2 +-
2 files changed, 2 insertions(+), 2 deletions(-)
diff --git a/site2/docs/io-quickstart.md b/site2/docs/io-quickstart.md
index ceefdeb..042ce8e 100644
--- a/site2/docs/io-quickstart.md
+++ b/site2/docs/io-quickstart.md
@@ -66,7 +66,7 @@ $ tar xvfz
/path/to/apache-pulsar-io-connectors-{{pulsar:version}}-bin.tar.gz
// you will find a directory named
`apache-pulsar-io-connectors-{{pulsar:version}}` in the pulsar directory
// then copy the connectors
-$ cd apache-pulsar-io-connectors-{{pulsar:version}}/connectors connectors
+$ cp -r apache-pulsar-io-connectors-{{pulsar:version}}/connectors connectors
$ ls connectors
pulsar-io-aerospike-{{pulsar.version}}.nar
diff --git
a/site2/website/versioned_docs/version-2.1.0-incubating/io-quickstart.md
b/site2/website/versioned_docs/version-2.1.0-incubating/io-quickstart.md
index 37868db..069e2bb 100644
--- a/site2/website/versioned_docs/version-2.1.0-incubating/io-quickstart.md
+++ b/site2/website/versioned_docs/version-2.1.0-incubating/io-quickstart.md
@@ -67,7 +67,7 @@ $ tar xvfz
/path/to/apache-pulsar-io-connectors-{{pulsar:version}}-bin.tar.gz
// you will find a directory named
`apache-pulsar-io-connectors-{{pulsar:version}}` in the pulsar directory
// then copy the connectors
-$ cd apache-pulsar-io-connectors-{{pulsar:version}}/connectors connectors
+$ cp -r apache-pulsar-io-connectors-{{pulsar:version}}/connectors connectors
$ ls connectors
pulsar-io-aerospike-{{pulsar.version}}.nar
[incubator-pulsar] branch master updated: Add support to restart function (#2365)
This is an automated email from the ASF dual-hosted git repository.
rdhabalia pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 7bcd893 Add support to restart function (#2365)
7bcd893 is described below
commit 7bcd8934a0a53ab7a62b3c9d77fbdec94ab497d2
Author: Rajan Dhabalia
AuthorDate: Mon Aug 13 22:49:11 2018 -0700
Add support to restart function (#2365)
* Add support to restart function
fix: pulsar function restart
* add support to restart all function instances
---
.../pulsar/broker/admin/impl/FunctionsBase.java| 25 +
.../apache/pulsar/io/PulsarFunctionTlsTest.java| 2 +-
.../org/apache/pulsar/io/PulsarSinkE2ETest.java| 63 +++-
.../org/apache/pulsar/client/admin/Functions.java | 33 +++
.../client/admin/internal/FunctionsImpl.java | 22 +
.../apache/pulsar/admin/cli/CmdFunctionsTest.java | 29 ++
.../org/apache/pulsar/admin/cli/CmdFunctions.java | 37 ++-
.../pulsar/functions/worker/FunctionActioner.java | 4 +-
.../functions/worker/FunctionRuntimeManager.java | 107 +
.../functions/worker/rest/api/FunctionsImpl.java | 66 +
.../worker/rest/api/v2/FunctionApiV2Resource.java | 28 ++
11 files changed, 404 insertions(+), 12 deletions(-)
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/FunctionsBase.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/FunctionsBase.java
index 564eb18..b8891e5 100644
---
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/FunctionsBase.java
+++
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/FunctionsBase.java
@@ -268,6 +268,31 @@ public class FunctionsBase extends AdminResource
implements Supplier typeArg = byte[].class;
FunctionDetails.Builder functionDetailsBuilder =
FunctionDetails.newBuilder();
@@ -446,7 +446,7 @@ public class PulsarSinkE2ETest {
String jarFilePathUrl = Utils.FILE + ":"
+
PulsarSink.class.getProtectionDomain().getCodeSource().getLocation().getPath();
FunctionDetails functionDetails = createSinkConfig(jarFilePathUrl,
tenant, namespacePortion, functionName,
-sinkTopic, subscriptionName);
+"my.*", sinkTopic, subscriptionName);
try {
admin.functions().createFunctionWithUrl(functionDetails,
jarFilePathUrl);
assertTrue(validRoleName);
@@ -507,4 +507,57 @@ public class PulsarSinkE2ETest {
assertEquals(functionMetadata.getSink().getTypeClassName(),
typeArgs[1].getName());
}
+
+@Test(timeOut = 2)
+public void testFunctionRestartApi() throws Exception {
+
+final String namespacePortion = "io";
+final String replNamespace = tenant + "/" + namespacePortion;
+final String sourceTopicName = "restartFunction";
+final String sourceTopic = "persistent://" + replNamespace + "/" +
sourceTopicName;
+final String sinkTopic = "persistent://" + replNamespace + "/output";
+final String functionName = "PulsarSink-test";
+final String subscriptionName = "test-sub";
+admin.namespaces().createNamespace(replNamespace);
+Set clusters = Sets.newHashSet(Lists.newArrayList("use"));
+admin.namespaces().setNamespaceReplicationClusters(replNamespace,
clusters);
+
+// create source topic
+Producer producer =
pulsarClient.newProducer().topic(sourceTopic).create();
+
+String jarFilePathUrl = Utils.FILE + ":"
++
PulsarSink.class.getProtectionDomain().getCodeSource().getLocation().getPath();
+FunctionDetails functionDetails = createSinkConfig(jarFilePathUrl,
tenant, namespacePortion, functionName,
+sourceTopicName, sinkTopic, subscriptionName);
+admin.functions().createFunctionWithUrl(functionDetails,
jarFilePathUrl);
+
+retryStrategically((test) -> {
+try {
+SubscriptionStats subStats =
admin.topics().getStats(sourceTopic).subscriptions.get(subscriptionName);
+return subStats != null && subStats.consumers.size() == 1;
+} catch (PulsarAdminException e) {
+return false;
+}
+}, 5, 150);
+
+SubscriptionStats subStats =
admin.topics().getStats(sourceTopic).subscriptions.get(subscriptionName);
+assertEquals(subStats.consumers.size(), 1);
+
+// it should restart consumer : so, check if consumer came up again
after restarting function
+admin.functions().restartFunction(tenant, namespacePortion,
functionName);
+
+retryStrategically((test) -> {
+try {
+SubscriptionStats subStat =
admin.topics().getStats(sourceTopic).subscriptions.get(subscriptionName);
+return
[GitHub] rdhabalia closed pull request #2365: Add support to restart function
rdhabalia closed pull request #2365: Add support to restart function
URL: https://github.com/apache/incubator-pulsar/pull/2365
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/FunctionsBase.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/FunctionsBase.java
index 564eb18067..b8891e5a92 100644
---
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/FunctionsBase.java
+++
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/FunctionsBase.java
@@ -267,6 +267,31 @@ public Response triggerFunction(final @PathParam("tenant")
String tenant,
}
+@POST
+@ApiOperation(value = "Restart function instance", response = Void.class)
+@ApiResponses(value = { @ApiResponse(code = 400, message = "Invalid
request"),
+@ApiResponse(code = 404, message = "The function does not exist"),
+@ApiResponse(code = 500, message = "Internal server error") })
+@Path("/{tenant}/{namespace}/{functionName}/{instanceId}/restart")
+@Consumes(MediaType.APPLICATION_JSON)
+public Response restartFunction(final @PathParam("tenant") String tenant,
+final @PathParam("namespace") String namespace, final
@PathParam("functionName") String functionName,
+final @PathParam("instanceId") String instanceId) {
+return functions.restartFunctionInstance(tenant, namespace,
functionName, instanceId);
+}
+
+@POST
+@ApiOperation(value = "Restart all function instances", response =
Void.class)
+@ApiResponses(value = { @ApiResponse(code = 400, message = "Invalid
request"),
+@ApiResponse(code = 404, message = "The function does not exist"),
+@ApiResponse(code = 500, message = "Internal server error") })
+@Path("/{tenant}/{namespace}/{functionName}/restart")
+@Consumes(MediaType.APPLICATION_JSON)
+public Response restartFunction(final @PathParam("tenant") String tenant,
+final @PathParam("namespace") String namespace, final
@PathParam("functionName") String functionName) {
+return functions.restartFunctionInstances(tenant, namespace,
functionName);
+}
+
@POST
@ApiOperation(
value = "Uploads Pulsar Function file data",
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionTlsTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionTlsTest.java
index c57a8a0e59..2254626804 100644
---
a/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionTlsTest.java
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionTlsTest.java
@@ -206,7 +206,7 @@ public void testAuthorization() throws Exception {
String jarFilePathUrl = String.format("%s:%s", Utils.FILE,
PulsarSink.class.getProtectionDomain().getCodeSource().getLocation().getPath());
FunctionDetails functionDetails =
PulsarSinkE2ETest.createSinkConfig(jarFilePathUrl, tenant, namespacePortion,
-functionName, sinkTopic, subscriptionName);
+functionName, "my.*", sinkTopic, subscriptionName);
try {
functionAdmin.functions().createFunctionWithUrl(functionDetails,
jarFilePathUrl);
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarSinkE2ETest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarSinkE2ETest.java
index abc4735ea1..e5902d4d00 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarSinkE2ETest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarSinkE2ETest.java
@@ -270,7 +270,7 @@ public void testE2EPulsarSink() throws Exception {
String jarFilePathUrl = Utils.FILE + ":"
+
PulsarSink.class.getProtectionDomain().getCodeSource().getLocation().getPath();
FunctionDetails functionDetails = createSinkConfig(jarFilePathUrl,
tenant, namespacePortion, functionName,
-sinkTopic, subscriptionName);
+"my.*", sinkTopic, subscriptionName);
admin.functions().createFunctionWithUrl(functionDetails,
jarFilePathUrl);
// try to update function to test: update-function functionality
@@ -332,7 +332,7 @@ public void testPulsarSinkStats() throws Exception {
String jarFilePathUrl = Utils.FILE + ":"
+
PulsarSink.class.getProtectionDomain().getCodeSource().getLocation().getPath();
FunctionDetails functionDetails = createSinkConfig(jarFilePathUrl,
tenant, namespacePortion, functionName,
-sinkTopic, subscriptionName);
+"my.*", sinkTopic, subscriptionName);
admin.functions().createFunctionWithUrl(functionDetails,
jarFilePathUrl);
[GitHub] srkukarni opened a new pull request #2369: If --customSerdeInputs is specified, please update the right variable
srkukarni opened a new pull request #2369: If --customSerdeInputs is specified, please update the right variable URL: https://github.com/apache/incubator-pulsar/pull/2369 ### Motivation When we made the switch to get away from camel-case arguments for functions, we moved the camel case arguments as deprecated, so users can still use them but they are hidden. But that change missed to take --customSerdeInputs into account. This fixes that ommision ### Modifications Describe the modifications you've done. ### Result After your change, what will change. This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] rdhabalia commented on a change in pull request #2365: Add support to restart function
rdhabalia commented on a change in pull request #2365: Add support to restart
function
URL: https://github.com/apache/incubator-pulsar/pull/2365#discussion_r209829347
##
File path:
pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionRuntimeManager.java
##
@@ -317,6 +326,104 @@ public synchronized void
removeAssignments(Collection assignments) {
return functionStatus;
}
+public Response restartFunctionInstance(String tenant, String namespace,
String functionName, int instanceId) throws Exception {
+Assignment assignment = this.findAssignment(tenant, namespace,
functionName, instanceId);
+final String fullFunctionName = String.format("%s/%s/%s/%s", tenant,
namespace, functionName, instanceId);
+if (assignment == null) {
+return
Response.status(Status.BAD_REQUEST).type(MediaType.APPLICATION_JSON)
+.entity(new ErrorData(fullFunctionName + " doesn't
exist")).build();
+}
+
+final String assignedWorkerId = assignment.getWorkerId();
+final String workerId = this.workerConfig.getWorkerId();
+
+if (assignedWorkerId.equals(workerId)) {
+
restartFunction(Utils.getFullyQualifiedInstanceId(assignment.getInstance()));
+return Response.status(Status.OK).build();
+} else {
+// query other worker
+List workerInfoList =
this.membershipManager.getCurrentMembership();
+WorkerInfo workerInfo = null;
+for (WorkerInfo entry : workerInfoList) {
+if (assignment.getWorkerId().equals(entry.getWorkerId())) {
+workerInfo = entry;
+}
+}
+if (workerInfo == null) {
+return
Response.status(Status.BAD_REQUEST).type(MediaType.APPLICATION_JSON)
+.entity(new ErrorData(fullFunctionName + " has not
been assigned yet")).build();
+}
+
+URI redirect = null;
+final String redirectUrl =
String.format("http://%s:%d/admin/functions/%s/%s/%s/%d/restart";,
+workerInfo.getWorkerHostname(), workerInfo.getPort(),
tenant, namespace, functionName, instanceId);
+try {
+redirect = new URI(redirectUrl);
+} catch (URISyntaxException e) {
+log.error("Error in preparing redirect url for {}/{}/{}/{}:
{}", tenant, namespace, functionName,
+instanceId, e.getMessage(), e);
+return
Response.status(Status.INTERNAL_SERVER_ERROR).type(MediaType.APPLICATION_JSON)
+.entity(new ErrorData(fullFunctionName + " invalid
redirection url")).build();
+}
+throw new
WebApplicationException(Response.temporaryRedirect(redirect).build());
+}
+}
+
+public Response restartFunctionInstances(String tenant, String namespace,
String functionName) throws Exception {
+final String fullFunctionName = String.format("%s/%s/%s", tenant,
namespace, functionName);
+Collection assignments =
this.findFunctionAssignments(tenant, namespace, functionName);
+
+if (assignments.isEmpty()) {
+return
Response.status(Status.BAD_REQUEST).type(MediaType.APPLICATION_JSON)
+.entity(new ErrorData(fullFunctionName + " has not been
assigned yet")).build();
+}
+for (Assignment assignment : assignments) {
+final String assignedWorkerId = assignment.getWorkerId();
Review comment:
actually if function owns by same worker then I created
`restartFunction(fullyQualifiedInstanceId);` to reuse code. but if function
owner is different then we need a blocking call to aggregate result. however,
in restartInstance function, we just redirect request to other host without
blocking call. so, we can't reuse redirection part.
Also, I am going to create separate PR for line 400 to use pulsar-admin to
reuse Web-Client and support proper auth&auth.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] k2la commented on issue #2368: Fix Copy Command in Pulsar IO Getting Started
k2la commented on issue #2368: Fix Copy Command in Pulsar IO Getting Started URL: https://github.com/apache/incubator-pulsar/pull/2368#issuecomment-412753369 retest this please This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] srkukarni commented on a change in pull request #2266: Integrate functions and io with schema registry
srkukarni commented on a change in pull request #2266: Integrate functions and
io with schema registry
URL: https://github.com/apache/incubator-pulsar/pull/2266#discussion_r209824852
##
File path:
pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSinks.java
##
@@ -204,14 +207,27 @@ void runCmd() throws Exception {
@Parameter(names = { "-t", "--sink-type" }, description = "The sinks's
connector provider")
protected String sinkType;
-@Parameter(names = "--inputs", description = "The sink's input topic
or topics (multiple topics can be specified as a comma-separated list)")
+@Parameter(names = { "-i",
+"--inputs" }, description = "The sink's input topic or topics
(multiple topics can be specified as a comma-separated list)")
protected String inputs;
-@Parameter(names = "--topicsPattern", description = "TopicsPattern to
consume from list of topics under a namespace that match the pattern. [--input]
and [--topicsPattern] are mutually exclusive. Add SerDe class name for a
pattern in --customSerdeInputs (supported for java fun only)")
+
+@Parameter(names = "--topicsPattern", description = "TopicsPattern to
consume from list of topics under a namespace that match the pattern. [--input]
and [--topicsPattern] are mutually exclusive. Add SerDe class name for a
pattern in --customSerdeInputs (supported for java fun only)", hidden = true)
protected String topicsPattern;
+
+@Parameter(names = { "-st",
+"--schema-type" }, description = "The builtin schema type (eg:
'avro', 'json', etc..) or the class name for a Schema or Serde implementation")
+protected String schemaTypeOrClassName = "";
+
@Parameter(names = "--subsName", description = "Pulsar source
subscription name if user wants a specific subscription-name for input-topic
consumer")
protected String subsName;
-@Parameter(names = "--customSerdeInputs", description = "The map of
input topics to SerDe class names (as a JSON string)")
+
+@Parameter(names = "--customSerdeInputs", description = "The map of
input topics to SerDe class names (as a JSON string)", hidden = true)
protected String customSerdeInputString;
Review comment:
Thats the style that we have followed elsewhere in CmdFunctions, so might be
useful to follow the same. Will be easier to clean these things later.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[incubator-pulsar] branch master updated: Set correct exception if function runnable fails with Error (#2353)
This is an automated email from the ASF dual-hosted git repository.
rdhabalia pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new c9988cf Set correct exception if function runnable fails with Error
(#2353)
c9988cf is described below
commit c9988cff24aa12e94c2a09cd112ab2bfa0a8860b
Author: Rajan Dhabalia
AuthorDate: Mon Aug 13 21:20:02 2018 -0700
Set correct exception if function runnable fails with Error (#2353)
* Set correct exception if function runnable fails with Error
* fix throwable set
---
.../org/apache/pulsar/functions/instance/JavaInstanceRunnable.java | 6 --
1 file changed, 4 insertions(+), 2 deletions(-)
diff --git
a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
index aea9d53..fe94b17 100644
---
a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
+++
b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
@@ -185,8 +185,10 @@ public class JavaInstanceRunnable implements
AutoCloseable, Runnable {
*/
@Override
public void run() {
+String functionName = null;
try {
ContextImpl contextImpl = setupContext();
+functionName = String.format("%s-%s", contextImpl.getTenant(),
contextImpl.getFunctionName());
javaInstance = setupJavaInstance(contextImpl);
if (null != stateTable) {
StateContextImpl stateContext = new
StateContextImpl(stateTable);
@@ -233,8 +235,8 @@ public class JavaInstanceRunnable implements AutoCloseable,
Runnable {
}
}
} catch (Throwable t) {
-log.error("Uncaught exception in Java Instance", t);
-deathException = (Exception) t;
+log.error("[{}] Uncaught exception in Java Instance",
functionName, t);
+deathException = t;
return;
} finally {
log.info("Closing instance");
[GitHub] srkukarni commented on a change in pull request #2365: Add support to restart function
srkukarni commented on a change in pull request #2365: Add support to restart
function
URL: https://github.com/apache/incubator-pulsar/pull/2365#discussion_r209824676
##
File path:
pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionRuntimeManager.java
##
@@ -317,6 +326,104 @@ public synchronized void
removeAssignments(Collection assignments) {
return functionStatus;
}
+public Response restartFunctionInstance(String tenant, String namespace,
String functionName, int instanceId) throws Exception {
+Assignment assignment = this.findAssignment(tenant, namespace,
functionName, instanceId);
+final String fullFunctionName = String.format("%s/%s/%s/%s", tenant,
namespace, functionName, instanceId);
+if (assignment == null) {
+return
Response.status(Status.BAD_REQUEST).type(MediaType.APPLICATION_JSON)
+.entity(new ErrorData(fullFunctionName + " doesn't
exist")).build();
+}
+
+final String assignedWorkerId = assignment.getWorkerId();
+final String workerId = this.workerConfig.getWorkerId();
+
+if (assignedWorkerId.equals(workerId)) {
+
restartFunction(Utils.getFullyQualifiedInstanceId(assignment.getInstance()));
+return Response.status(Status.OK).build();
+} else {
+// query other worker
+List workerInfoList =
this.membershipManager.getCurrentMembership();
+WorkerInfo workerInfo = null;
+for (WorkerInfo entry : workerInfoList) {
+if (assignment.getWorkerId().equals(entry.getWorkerId())) {
+workerInfo = entry;
+}
+}
+if (workerInfo == null) {
+return
Response.status(Status.BAD_REQUEST).type(MediaType.APPLICATION_JSON)
+.entity(new ErrorData(fullFunctionName + " has not
been assigned yet")).build();
+}
+
+URI redirect = null;
+final String redirectUrl =
String.format("http://%s:%d/admin/functions/%s/%s/%s/%d/restart";,
+workerInfo.getWorkerHostname(), workerInfo.getPort(),
tenant, namespace, functionName, instanceId);
+try {
+redirect = new URI(redirectUrl);
+} catch (URISyntaxException e) {
+log.error("Error in preparing redirect url for {}/{}/{}/{}:
{}", tenant, namespace, functionName,
+instanceId, e.getMessage(), e);
+return
Response.status(Status.INTERNAL_SERVER_ERROR).type(MediaType.APPLICATION_JSON)
+.entity(new ErrorData(fullFunctionName + " invalid
redirection url")).build();
+}
+throw new
WebApplicationException(Response.temporaryRedirect(redirect).build());
+}
+}
+
+public Response restartFunctionInstances(String tenant, String namespace,
String functionName) throws Exception {
+final String fullFunctionName = String.format("%s/%s/%s", tenant,
namespace, functionName);
+Collection assignments =
this.findFunctionAssignments(tenant, namespace, functionName);
+
+if (assignments.isEmpty()) {
+return
Response.status(Status.BAD_REQUEST).type(MediaType.APPLICATION_JSON)
+.entity(new ErrorData(fullFunctionName + " has not been
assigned yet")).build();
+}
+for (Assignment assignment : assignments) {
+final String assignedWorkerId = assignment.getWorkerId();
Review comment:
a good refactor could be to reuse above function?
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] rdhabalia closed pull request #2353: Set correct exception if function runnable fails with Error
rdhabalia closed pull request #2353: Set correct exception if function runnable
fails with Error
URL: https://github.com/apache/incubator-pulsar/pull/2353
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
index aea9d53d93..fe94b179f8 100644
---
a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
+++
b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
@@ -185,8 +185,10 @@ ContextImpl setupContext() {
*/
@Override
public void run() {
+String functionName = null;
try {
ContextImpl contextImpl = setupContext();
+functionName = String.format("%s-%s", contextImpl.getTenant(),
contextImpl.getFunctionName());
javaInstance = setupJavaInstance(contextImpl);
if (null != stateTable) {
StateContextImpl stateContext = new
StateContextImpl(stateTable);
@@ -233,8 +235,8 @@ public void run() {
}
}
} catch (Throwable t) {
-log.error("Uncaught exception in Java Instance", t);
-deathException = (Exception) t;
+log.error("[{}] Uncaught exception in Java Instance",
functionName, t);
+deathException = t;
return;
} finally {
log.info("Closing instance");
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] k2la opened a new pull request #2368: Fix Copy Command in Pulsar IO Getting Started
k2la opened a new pull request #2368: Fix Copy Command in Pulsar IO Getting Started URL: https://github.com/apache/incubator-pulsar/pull/2368 ### Motivation In https://pulsar.incubator.apache.org/docs/en/io-quickstart/ , `connectors` directory cannot be copied because copy command is wrong. > // you will find a directory named `apache-pulsar-io-connectors-2.1.0-incubating` in the pulsar directory > // then copy the connectors > > $ cd apache-pulsar-io-connectors-2.1.0-incubating/connectors connectors ### Modifications ・Change `cd` to `cp -r` ### Result `connectors` directory can be copied successfully. ``` $ cp -r apache-pulsar-io-connectors-2.1.0-incubating/connectors connectors $ ls connectors pulsar-io-aerospike-2.1.0-incubating.nar pulsar-io-kinesis-2.1.0-incubating.nar pulsar-io-cassandra-2.1.0-incubating.nar pulsar-io-rabbitmq-2.1.0-incubating.nar pulsar-io-kafka-2.1.0-incubating.nar pulsar-io-twitter-2.1.0-incubating.nar ``` This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] rdhabalia commented on issue #2365: Add support to restart function
rdhabalia commented on issue #2365: Add support to restart function URL: https://github.com/apache/incubator-pulsar/pull/2365#issuecomment-412728342 @srkukarni added support to restart all instances of the function. This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] rdhabalia commented on issue #2365: Add support to restart function
rdhabalia commented on issue #2365: Add support to restart function URL: https://github.com/apache/incubator-pulsar/pull/2365#issuecomment-412723540 > If one doesn't specify instance_id does it make sense to restart all instances? Sure, let me make the change. This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] sijie opened a new pull request #2367: [compaction] make topic compaction works with partitioned topic
sijie opened a new pull request #2367: [compaction] make topic compaction works with partitioned topic URL: https://github.com/apache/incubator-pulsar/pull/2367 ### Motivation Topic compaction doesn't work with partitioned topic. ### Changes - make `RawReaderImpl` and `ReaderImpl` return message with partition idx - make broker service `Consumer` deliver MessageIdData with partition idx - `MultiTopicsConsumerImpl` should use `-1` as partition idx for non-partitioned topic - add an integration test to ensure compaction work with partitioned topic ### Motivation Explain here the context, and why you're making that change. What is the problem you're trying to solve. ### Modifications Describe the modifications you've done. ### Result After your change, what will change. This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[incubator-pulsar] branch master updated: Fixed linking of python wrapper to boost-python in MacOS build (#2366)
This is an automated email from the ASF dual-hosted git repository.
hrsakai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 82771da Fixed linking of python wrapper to boost-python in MacOS
build (#2366)
82771da is described below
commit 82771daadd8d33eaf5087b43328458f970c6c378
Author: Matteo Merli
AuthorDate: Mon Aug 13 16:56:23 2018 -0700
Fixed linking of python wrapper to boost-python in MacOS build (#2366)
---
pulsar-client-cpp/python/CMakeLists.txt| 4 +++-
pulsar-client-cpp/python/pkg/osx/build.sh.template | 14 ++
2 files changed, 13 insertions(+), 5 deletions(-)
diff --git a/pulsar-client-cpp/python/CMakeLists.txt
b/pulsar-client-cpp/python/CMakeLists.txt
index 8e70f72..d506844 100644
--- a/pulsar-client-cpp/python/CMakeLists.txt
+++ b/pulsar-client-cpp/python/CMakeLists.txt
@@ -35,7 +35,9 @@ if("${CMAKE_CXX_COMPILER_ID}" STREQUAL "Clang")
set(CMAKE_SHARED_LIBRARY_CREATE_CXX_FLAGS
"${CMAKE_SHARED_LIBRARY_CREATE_CXX_FLAGS} -undefined dynamic_lookup")
endif()
-set(PYTHON_WRAPPER_LIBS ${Boost_PYTHON_LIBRARY} ${Boost_PYTHON3_LIBRARY})
+# Try all possible boost-python variable namings
+set(PYTHON_WRAPPER_LIBS ${Boost_PYTHON_LIBRARY} ${Boost_PYTHON3_LIBRARY}
+${Boost_PYTHON27-MT_LIBRARY}
${Boost_PYTHON37-MT_LIBRARY})
if (APPLE)
target_link_libraries(_pulsar -Wl,-all_load pulsarStatic
${PYTHON_WRAPPER_LIBS})
diff --git a/pulsar-client-cpp/python/pkg/osx/build.sh.template
b/pulsar-client-cpp/python/pkg/osx/build.sh.template
index 71687ac..40b7420 100755
--- a/pulsar-client-cpp/python/pkg/osx/build.sh.template
+++ b/pulsar-client-cpp/python/pkg/osx/build.sh.template
@@ -27,7 +27,7 @@ BOOST_VERSION=1.67
/usr/bin/ruby -e "$(curl -fsSL
https://raw.githubusercontent.com/Homebrew/install/master/install)"
-brew install openssl git boost@$BOOST_VERSION pkg-config jsoncpp cmake
protobuf260 log4cxx
+brew install openssl git boost@$BOOST_VERSION pkg-config jsoncpp cmake
protobuf260
brew link --force boost@$BOOST_VERSION
@@ -40,8 +40,8 @@ fi
if [ "$PYTHONVER" = "PYTHON3" ]
then
- brew install python boost-python3@BOOST_VERSION
- brew link --force boost-python3@$BOOST_VERSION
+ brew install python boost-python3
+ brew link --force boost-python3
fi
brew link --force protobuf260
@@ -52,4 +52,10 @@ cd incubator-pulsar/pulsar-client-cpp
cmake . -DBUILD_TESTS=OFF -DLINK_STATIC=ON
make _pulsar -j8
cd python
-python setup.py bdist_wheel
+
+if [ "$PYTHONVER" = "PYTHON2" ]
+then
+ python2 setup.py bdist_wheel
+else
+ python3 setup.py bdist_wheel
+fi
[GitHub] hrsakai closed pull request #2366: Fixed linking of python wrapper to boost-python in MacOS build
hrsakai closed pull request #2366: Fixed linking of python wrapper to
boost-python in MacOS build
URL: https://github.com/apache/incubator-pulsar/pull/2366
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/pulsar-client-cpp/python/CMakeLists.txt
b/pulsar-client-cpp/python/CMakeLists.txt
index 8e70f72ab4..d506844482 100644
--- a/pulsar-client-cpp/python/CMakeLists.txt
+++ b/pulsar-client-cpp/python/CMakeLists.txt
@@ -35,7 +35,9 @@ if("${CMAKE_CXX_COMPILER_ID}" STREQUAL "Clang")
set(CMAKE_SHARED_LIBRARY_CREATE_CXX_FLAGS
"${CMAKE_SHARED_LIBRARY_CREATE_CXX_FLAGS} -undefined dynamic_lookup")
endif()
-set(PYTHON_WRAPPER_LIBS ${Boost_PYTHON_LIBRARY} ${Boost_PYTHON3_LIBRARY})
+# Try all possible boost-python variable namings
+set(PYTHON_WRAPPER_LIBS ${Boost_PYTHON_LIBRARY} ${Boost_PYTHON3_LIBRARY}
+${Boost_PYTHON27-MT_LIBRARY}
${Boost_PYTHON37-MT_LIBRARY})
if (APPLE)
target_link_libraries(_pulsar -Wl,-all_load pulsarStatic
${PYTHON_WRAPPER_LIBS})
diff --git a/pulsar-client-cpp/python/pkg/osx/build.sh.template
b/pulsar-client-cpp/python/pkg/osx/build.sh.template
index 71687ac4b4..40b7420de8 100755
--- a/pulsar-client-cpp/python/pkg/osx/build.sh.template
+++ b/pulsar-client-cpp/python/pkg/osx/build.sh.template
@@ -27,7 +27,7 @@ BOOST_VERSION=1.67
/usr/bin/ruby -e "$(curl -fsSL
https://raw.githubusercontent.com/Homebrew/install/master/install)"
-brew install openssl git boost@$BOOST_VERSION pkg-config jsoncpp cmake
protobuf260 log4cxx
+brew install openssl git boost@$BOOST_VERSION pkg-config jsoncpp cmake
protobuf260
brew link --force boost@$BOOST_VERSION
@@ -40,8 +40,8 @@ fi
if [ "$PYTHONVER" = "PYTHON3" ]
then
- brew install python boost-python3@BOOST_VERSION
- brew link --force boost-python3@$BOOST_VERSION
+ brew install python boost-python3
+ brew link --force boost-python3
fi
brew link --force protobuf260
@@ -52,4 +52,10 @@ cd incubator-pulsar/pulsar-client-cpp
cmake . -DBUILD_TESTS=OFF -DLINK_STATIC=ON
make _pulsar -j8
cd python
-python setup.py bdist_wheel
+
+if [ "$PYTHONVER" = "PYTHON2" ]
+then
+ python2 setup.py bdist_wheel
+else
+ python3 setup.py bdist_wheel
+fi
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] merlimat commented on a change in pull request #2266: Integrate functions and io with schema registry
merlimat commented on a change in pull request #2266: Integrate functions and
io with schema registry
URL: https://github.com/apache/incubator-pulsar/pull/2266#discussion_r209789545
##
File path:
pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSinks.java
##
@@ -204,14 +207,27 @@ void runCmd() throws Exception {
@Parameter(names = { "-t", "--sink-type" }, description = "The sinks's
connector provider")
protected String sinkType;
-@Parameter(names = "--inputs", description = "The sink's input topic
or topics (multiple topics can be specified as a comma-separated list)")
+@Parameter(names = { "-i",
+"--inputs" }, description = "The sink's input topic or topics
(multiple topics can be specified as a comma-separated list)")
protected String inputs;
-@Parameter(names = "--topicsPattern", description = "TopicsPattern to
consume from list of topics under a namespace that match the pattern. [--input]
and [--topicsPattern] are mutually exclusive. Add SerDe class name for a
pattern in --customSerdeInputs (supported for java fun only)")
+
+@Parameter(names = "--topicsPattern", description = "TopicsPattern to
consume from list of topics under a namespace that match the pattern. [--input]
and [--topicsPattern] are mutually exclusive. Add SerDe class name for a
pattern in --customSerdeInputs (supported for java fun only)", hidden = true)
protected String topicsPattern;
+
+@Parameter(names = { "-st",
+"--schema-type" }, description = "The builtin schema type (eg:
'avro', 'json', etc..) or the class name for a Schema or Serde implementation")
+protected String schemaTypeOrClassName = "";
+
@Parameter(names = "--subsName", description = "Pulsar source
subscription name if user wants a specific subscription-name for input-topic
consumer")
protected String subsName;
-@Parameter(names = "--customSerdeInputs", description = "The map of
input topics to SerDe class names (as a JSON string)")
+
+@Parameter(names = "--customSerdeInputs", description = "The map of
input topics to SerDe class names (as a JSON string)", hidden = true)
protected String customSerdeInputString;
Review comment:
It's marked as `hidden` so it won't show up in the command help, but yes,
we can also rename the variable itself.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] merlimat opened a new pull request #2366: Fixed linking of python wrapper to boost-python in MacOS build
merlimat opened a new pull request #2366: Fixed linking of python wrapper to boost-python in MacOS build URL: https://github.com/apache/incubator-pulsar/pull/2366 ### Motivation With newer cmake and boost versions from Homebrew, the variable for boost-python library name path had slightly changed. Adjusting to have successful build in all macos/python variants. This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[incubator-pulsar] branch master updated: [cpp] receiver queue size config acorss partitions in multi-topics-consumer (#2311)
This is an automated email from the ASF dual-hosted git repository. mmerli pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/incubator-pulsar.git The following commit(s) were added to refs/heads/master by this push: new 296830a [cpp] receiver queue size config acorss partitions in multi-topics-consumer (#2311) 296830a is described below commit 296830a37d82de65d30dfd74d9a5893033056f18 Author: Jia Zhai AuthorDate: Tue Aug 14 06:05:59 2018 +0800 [cpp] receiver queue size config acorss partitions in multi-topics-consumer (#2311) * catch up receiver queue size support in multi topics consumer * add python config --- pulsar-client-cpp/lib/MultiTopicsConsumerImpl.cc | 5 - pulsar-client-cpp/python/pulsar/__init__.py| 13 pulsar-client-cpp/python/src/config.cc | 4 .../client/impl/MultiTopicsConsumerImpl.java | 23 +- 4 files changed, 35 insertions(+), 10 deletions(-) diff --git a/pulsar-client-cpp/lib/MultiTopicsConsumerImpl.cc b/pulsar-client-cpp/lib/MultiTopicsConsumerImpl.cc index 7be197c..6750273 100644 --- a/pulsar-client-cpp/lib/MultiTopicsConsumerImpl.cc +++ b/pulsar-client-cpp/lib/MultiTopicsConsumerImpl.cc @@ -160,9 +160,12 @@ void MultiTopicsConsumerImpl::subscribeTopicPartitions(const Result result, config.setBrokerConsumerStatsCacheTimeInMs(conf_.getBrokerConsumerStatsCacheTimeInMs()); config.setMessageListener( boost::bind(&MultiTopicsConsumerImpl::messageReceived, shared_from_this(), _1, _2)); -config.setReceiverQueueSize(conf_.getReceiverQueueSize()); int numPartitions = partitionMetadata->getPartitions() >= 1 ? partitionMetadata->getPartitions() : 1; +// Apply total limit of receiver queue size across partitions +config.setReceiverQueueSize( +std::min(conf_.getReceiverQueueSize(), + (int)(conf_.getMaxTotalReceiverQueueSizeAcrossPartitions() / numPartitions))); Lock lock(mutex_); topicsPartitions_.insert(std::make_pair(topicName->toString(), numPartitions)); diff --git a/pulsar-client-cpp/python/pulsar/__init__.py b/pulsar-client-cpp/python/pulsar/__init__.py index 434fb07..f3b560b 100644 --- a/pulsar-client-cpp/python/pulsar/__init__.py +++ b/pulsar-client-cpp/python/pulsar/__init__.py @@ -315,6 +315,7 @@ class Client: send_timeout_millis=3, compression_type=CompressionType.NONE, max_pending_messages=1000, +max_pending_messages_across_partitions=5, block_if_queue_full=False, batching_enabled=False, batching_max_messages=1000, @@ -352,6 +353,9 @@ class Client: * `max_pending_messages`: Set the max size of the queue holding the messages pending to receive an acknowledgment from the broker. +* `max_pending_messages_across_partitions`: + Set the max size of the queue holding the messages pending to receive + an acknowledgment across partitions from the broker. * `block_if_queue_full`: Set whether `send_async` operations should block when the outgoing message queue is full. * `message_routing_mode`: @@ -364,6 +368,7 @@ class Client: _check_type(int, send_timeout_millis, 'send_timeout_millis') _check_type(CompressionType, compression_type, 'compression_type') _check_type(int, max_pending_messages, 'max_pending_messages') +_check_type(int, max_pending_messages_across_partitions, 'max_pending_messages_across_partitions') _check_type(bool, block_if_queue_full, 'block_if_queue_full') _check_type(bool, batching_enabled, 'batching_enabled') _check_type(int, batching_max_messages, 'batching_max_messages') @@ -374,6 +379,7 @@ class Client: conf.send_timeout_millis(send_timeout_millis) conf.compression_type(compression_type) conf.max_pending_messages(max_pending_messages) + conf.max_pending_messages_across_partitions(max_pending_messages_across_partitions) conf.block_if_queue_full(block_if_queue_full) conf.batching_enabled(batching_enabled) conf.batching_max_messages(batching_max_messages) @@ -392,6 +398,7 @@ class Client: consumer_type=ConsumerType.Exclusive, message_listener=None, receiver_queue_size=1000, + max_total_receiver_queue_size_across_partitions=5, consumer_name=None, unacked_messages_timeout_ms=None, broker_consumer_stats_cache_time_ms=3, @@ -434,6 +441,9 @@ class Client: should not be interrupted when the consumer queue size is zero. The default value is 1000 messages and should work well for most use cases. +* `max
[GitHub] merlimat closed pull request #2311: [cpp] receiver queue size config acorss partitions in multi-topics-consumer
merlimat closed pull request #2311: [cpp] receiver queue size config acorss partitions in multi-topics-consumer URL: https://github.com/apache/incubator-pulsar/pull/2311 This is a PR merged from a forked repository. As GitHub hides the original diff on merge, it is displayed below for the sake of provenance: As this is a foreign pull request (from a fork), the diff is supplied below (as it won't show otherwise due to GitHub magic): diff --git a/pulsar-client-cpp/lib/MultiTopicsConsumerImpl.cc b/pulsar-client-cpp/lib/MultiTopicsConsumerImpl.cc index 7be197c89c..6750273649 100644 --- a/pulsar-client-cpp/lib/MultiTopicsConsumerImpl.cc +++ b/pulsar-client-cpp/lib/MultiTopicsConsumerImpl.cc @@ -160,9 +160,12 @@ void MultiTopicsConsumerImpl::subscribeTopicPartitions(const Result result, config.setBrokerConsumerStatsCacheTimeInMs(conf_.getBrokerConsumerStatsCacheTimeInMs()); config.setMessageListener( boost::bind(&MultiTopicsConsumerImpl::messageReceived, shared_from_this(), _1, _2)); -config.setReceiverQueueSize(conf_.getReceiverQueueSize()); int numPartitions = partitionMetadata->getPartitions() >= 1 ? partitionMetadata->getPartitions() : 1; +// Apply total limit of receiver queue size across partitions +config.setReceiverQueueSize( +std::min(conf_.getReceiverQueueSize(), + (int)(conf_.getMaxTotalReceiverQueueSizeAcrossPartitions() / numPartitions))); Lock lock(mutex_); topicsPartitions_.insert(std::make_pair(topicName->toString(), numPartitions)); diff --git a/pulsar-client-cpp/python/pulsar/__init__.py b/pulsar-client-cpp/python/pulsar/__init__.py index 434fb07a41..f3b560b747 100644 --- a/pulsar-client-cpp/python/pulsar/__init__.py +++ b/pulsar-client-cpp/python/pulsar/__init__.py @@ -315,6 +315,7 @@ def create_producer(self, topic, send_timeout_millis=3, compression_type=CompressionType.NONE, max_pending_messages=1000, +max_pending_messages_across_partitions=5, block_if_queue_full=False, batching_enabled=False, batching_max_messages=1000, @@ -352,6 +353,9 @@ def create_producer(self, topic, * `max_pending_messages`: Set the max size of the queue holding the messages pending to receive an acknowledgment from the broker. +* `max_pending_messages_across_partitions`: + Set the max size of the queue holding the messages pending to receive + an acknowledgment across partitions from the broker. * `block_if_queue_full`: Set whether `send_async` operations should block when the outgoing message queue is full. * `message_routing_mode`: @@ -364,6 +368,7 @@ def create_producer(self, topic, _check_type(int, send_timeout_millis, 'send_timeout_millis') _check_type(CompressionType, compression_type, 'compression_type') _check_type(int, max_pending_messages, 'max_pending_messages') +_check_type(int, max_pending_messages_across_partitions, 'max_pending_messages_across_partitions') _check_type(bool, block_if_queue_full, 'block_if_queue_full') _check_type(bool, batching_enabled, 'batching_enabled') _check_type(int, batching_max_messages, 'batching_max_messages') @@ -374,6 +379,7 @@ def create_producer(self, topic, conf.send_timeout_millis(send_timeout_millis) conf.compression_type(compression_type) conf.max_pending_messages(max_pending_messages) + conf.max_pending_messages_across_partitions(max_pending_messages_across_partitions) conf.block_if_queue_full(block_if_queue_full) conf.batching_enabled(batching_enabled) conf.batching_max_messages(batching_max_messages) @@ -392,6 +398,7 @@ def subscribe(self, topic, subscription_name, consumer_type=ConsumerType.Exclusive, message_listener=None, receiver_queue_size=1000, + max_total_receiver_queue_size_across_partitions=5, consumer_name=None, unacked_messages_timeout_ms=None, broker_consumer_stats_cache_time_ms=3, @@ -434,6 +441,9 @@ def my_listener(consumer, message): should not be interrupted when the consumer queue size is zero. The default value is 1000 messages and should work well for most use cases. +* `max_total_receiver_queue_size_across_partitions` + Set the max total receiver queue size across partitions. + This setting will be used to reduce the receiver queue size for individual partitions * `consumer_name`: Sets the consumer name. * `unacked_messages_timeout_ms`: @@ -450,6 +460,8 @@ def my_listener(consumer, message): _check_type(str, subscription_name, 'subscri
[GitHub] srkukarni commented on a change in pull request #2266: Integrate functions and io with schema registry
srkukarni commented on a change in pull request #2266: Integrate functions and
io with schema registry
URL: https://github.com/apache/incubator-pulsar/pull/2266#discussion_r209770637
##
File path: pulsar-functions/proto/src/main/proto/Function.proto
##
@@ -68,7 +73,15 @@ message SourceSpec {
// configs used only when source feeds into functions
SubscriptionType subscriptionType = 3;
+
+ // @deprecated -- use topicsToSchema
Review comment:
why not mark explicitly as deprecated in proto syntax?
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] sijie closed pull request #2354: add function package-url into function-get api response
sijie closed pull request #2354: add function package-url into function-get api
response
URL: https://github.com/apache/incubator-pulsar/pull/2354
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/pulsar-functions/proto/src/main/proto/Function.proto
b/pulsar-functions/proto/src/main/proto/Function.proto
index 92f14f6f19..77a9c2c0b6 100644
--- a/pulsar-functions/proto/src/main/proto/Function.proto
+++ b/pulsar-functions/proto/src/main/proto/Function.proto
@@ -58,6 +58,7 @@ message FunctionDetails {
SourceSpec source = 11;
SinkSpec sink = 12;
Resources resources = 13;
+string packageUrl = 14; //present only if function submitted with
package-url
}
message SourceSpec {
diff --git
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java
index c35dd52e80..a697363719 100644
---
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java
+++
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java
@@ -732,9 +732,8 @@ private FunctionDetails
validateUpdateRequestParamsWithPkgUrl(String tenant, Str
throw new IllegalArgumentException("Function Package url is not
valid. supported url (http/https/file)");
}
Utils.validateFileUrl(functionPkgUrl,
workerServiceSupplier.get().getWorkerConfig().getDownloadDirectory());
-File jarWithFileUrl = functionPkgUrl.startsWith(FILE) ? (new File((new
URL(functionPkgUrl)).toURI())) : null;
FunctionDetails functionDetails = validateUpdateRequestParams(tenant,
namespace, functionName,
-functionDetailsJson, jarWithFileUrl);
+functionDetailsJson, functionPkgUrl);
return functionDetails;
}
@@ -788,7 +787,7 @@ private String getFunctionCodeBuiltin(FunctionDetails
functionDetails) {
}
private FunctionDetails validateUpdateRequestParams(String tenant, String
namespace, String functionName,
-String functionDetailsJson, File jarWithFileUrl) throws
IllegalArgumentException {
+String functionDetailsJson, String functionPkgUrl) throws
IllegalArgumentException {
if (tenant == null) {
throw new IllegalArgumentException("Tenant is not provided");
}
@@ -805,7 +804,14 @@ private FunctionDetails validateUpdateRequestParams(String
tenant, String namesp
try {
FunctionDetails.Builder functionDetailsBuilder =
FunctionDetails.newBuilder();
org.apache.pulsar.functions.utils.Utils.mergeJson(functionDetailsJson,
functionDetailsBuilder);
-validateFunctionClassTypes(jarWithFileUrl, functionDetailsBuilder);
+if (isNotBlank(functionPkgUrl)) {
+// validate function details by loading function-jar from
local file-system
+File jarWithFileUrl = functionPkgUrl.startsWith(FILE) ? (new
File((new URL(functionPkgUrl)).toURI()))
+: null;
+validateFunctionClassTypes(jarWithFileUrl,
functionDetailsBuilder);
+// set package-url if present
+functionDetailsBuilder.setPackageUrl(functionPkgUrl);
+}
FunctionDetails functionDetails = functionDetailsBuilder.build();
List missingFields = new LinkedList<>();
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[incubator-pulsar] branch master updated: add function package-url into function-get api response (#2354)
This is an automated email from the ASF dual-hosted git repository.
sijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new cb2f4ee add function package-url into function-get api response
(#2354)
cb2f4ee is described below
commit cb2f4ee3960fc841ad42bc6e6b1209a92b88d5d9
Author: Rajan Dhabalia
AuthorDate: Mon Aug 13 14:40:18 2018 -0700
add function package-url into function-get api response (#2354)
### Motivation
Right now, we mostly use function by submitting function with url and many
times, we want to verify function package-url to confirm the package-version.
So, function-admin api should return packageurl along with function metadata.
### Modifications
return function pacakge-url along with function-metadata on get-function
api call.
---
pulsar-functions/proto/src/main/proto/Function.proto | 1 +
.../pulsar/functions/worker/rest/api/FunctionsImpl.java| 14 ++
2 files changed, 11 insertions(+), 4 deletions(-)
diff --git a/pulsar-functions/proto/src/main/proto/Function.proto
b/pulsar-functions/proto/src/main/proto/Function.proto
index 92f14f6..77a9c2c 100644
--- a/pulsar-functions/proto/src/main/proto/Function.proto
+++ b/pulsar-functions/proto/src/main/proto/Function.proto
@@ -58,6 +58,7 @@ message FunctionDetails {
SourceSpec source = 11;
SinkSpec sink = 12;
Resources resources = 13;
+string packageUrl = 14; //present only if function submitted with
package-url
}
message SourceSpec {
diff --git
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java
index b71e6cd..4bb6e49 100644
---
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java
+++
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java
@@ -733,9 +733,8 @@ public class FunctionsImpl {
throw new IllegalArgumentException("Function Package url is not
valid. supported url (http/https/file)");
}
Utils.validateFileUrl(functionPkgUrl,
workerServiceSupplier.get().getWorkerConfig().getDownloadDirectory());
-File jarWithFileUrl = functionPkgUrl.startsWith(FILE) ? (new File((new
URL(functionPkgUrl)).toURI())) : null;
FunctionDetails functionDetails = validateUpdateRequestParams(tenant,
namespace, functionName,
-functionDetailsJson, jarWithFileUrl);
+functionDetailsJson, functionPkgUrl);
return functionDetails;
}
@@ -789,7 +788,7 @@ public class FunctionsImpl {
}
private FunctionDetails validateUpdateRequestParams(String tenant, String
namespace, String functionName,
-String functionDetailsJson, File jarWithFileUrl) throws
IllegalArgumentException {
+String functionDetailsJson, String functionPkgUrl) throws
IllegalArgumentException {
if (tenant == null) {
throw new IllegalArgumentException("Tenant is not provided");
}
@@ -806,7 +805,14 @@ public class FunctionsImpl {
try {
FunctionDetails.Builder functionDetailsBuilder =
FunctionDetails.newBuilder();
org.apache.pulsar.functions.utils.Utils.mergeJson(functionDetailsJson,
functionDetailsBuilder);
-validateFunctionClassTypes(jarWithFileUrl, functionDetailsBuilder);
+if (isNotBlank(functionPkgUrl)) {
+// validate function details by loading function-jar from
local file-system
+File jarWithFileUrl = functionPkgUrl.startsWith(FILE) ? (new
File((new URL(functionPkgUrl)).toURI()))
+: null;
+validateFunctionClassTypes(jarWithFileUrl,
functionDetailsBuilder);
+// set package-url if present
+functionDetailsBuilder.setPackageUrl(functionPkgUrl);
+}
FunctionDetails functionDetails = functionDetailsBuilder.build();
List missingFields = new LinkedList<>();
[GitHub] srkukarni commented on a change in pull request #2266: Integrate functions and io with schema registry
srkukarni commented on a change in pull request #2266: Integrate functions and
io with schema registry
URL: https://github.com/apache/incubator-pulsar/pull/2266#discussion_r209768156
##
File path:
pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java
##
@@ -30,59 +30,60 @@
import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.ConsumerEventListener;
-import org.apache.pulsar.client.api.Message;
-import org.apache.pulsar.client.api.MessageBuilder;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.TypedMessageBuilder;
import org.apache.pulsar.functions.api.Record;
-import org.apache.pulsar.functions.api.SerDe;
-import org.apache.pulsar.functions.api.utils.DefaultSerDe;
-import org.apache.pulsar.functions.instance.InstanceUtils;
import org.apache.pulsar.functions.instance.SinkRecord;
import
org.apache.pulsar.functions.instance.producers.AbstractOneOuputTopicProducers;
import
org.apache.pulsar.functions.instance.producers.MultiConsumersOneOuputTopicProducers;
import org.apache.pulsar.functions.instance.producers.Producers;
import org.apache.pulsar.functions.source.PulsarRecord;
+import org.apache.pulsar.functions.source.TopicSchema;
import org.apache.pulsar.functions.utils.FunctionConfig;
import org.apache.pulsar.functions.utils.Reflections;
import org.apache.pulsar.io.core.Sink;
import org.apache.pulsar.io.core.SinkContext;
-import net.jodah.typetools.TypeResolver;
-
@Slf4j
public class PulsarSink implements Sink {
-private PulsarClient client;
-private PulsarSinkConfig pulsarSinkConfig;
-private SerDe outputSerDe;
+private final PulsarClient client;
+private final PulsarSinkConfig pulsarSinkConfig;
+
+private PulsarSinkProcessor pulsarSinkProcessor;
-private PulsarSinkProcessor pulsarSinkProcessor;
+private final TopicSchema topicSchema;
private interface PulsarSinkProcessor {
-void initializeOutputProducer(String outputTopic) throws Exception;
+void initializeOutputProducer(String outputTopic, Schema schema)
throws Exception;
Review comment:
Some comments around these?
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] sijie commented on issue #2353: Set correct exception if function runnable fails with Error
sijie commented on issue #2353: Set correct exception if function runnable fails with Error URL: https://github.com/apache/incubator-pulsar/pull/2353#issuecomment-412670139 run java8 tests This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] rdhabalia commented on issue #2347: Fix: function with multi-topic not acking on effectively-once
rdhabalia commented on issue #2347: Fix: function with multi-topic not acking on effectively-once URL: https://github.com/apache/incubator-pulsar/pull/2347#issuecomment-412669097 sure.. 👍 This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] sijie commented on issue #2347: Fix: function with multi-topic not acking on effectively-once
sijie commented on issue #2347: Fix: function with multi-topic not acking on effectively-once URL: https://github.com/apache/incubator-pulsar/pull/2347#issuecomment-412668786 @rdhabalia merged. I also changed the milestone to 2.2.0 since it is merged to master. when it is cherry-picked to 2.1.1, will update the milestone. otherwise, let's assume the changes based on master will be parted to major releases. This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[incubator-pulsar] branch master updated: Fix: function with multi-topic not acking on effectively-once (#2347)
This is an automated email from the ASF dual-hosted git repository.
sijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 0d2154e Fix: function with multi-topic not acking on effectively-once
(#2347)
0d2154e is described below
commit 0d2154ed3acd67bcb651644ae13e43cdd4045e90
Author: Rajan Dhabalia
AuthorDate: Mon Aug 13 14:19:46 2018 -0700
Fix: function with multi-topic not acking on effectively-once (#2347)
### Motivation
`MultiTopicsConsumerImpl` doesn't support `acknowledgeCumulativeAsync` and
therefore, function with multi-topic and `EFFECTIVELY_ONCE` processing is not
acking message and failing `EFFECTIVELY_ONCE` behavior.
### Modifications
Function should ack message for a specific topic consumer if
`inputTopicConsumer` is multi-topic consumer.
### Result
Function should able to ack messages for multi-topic consumer when
processing-guarantee is `EFFECTIVELY_ONCE`
---
.../client/api/PartitionedProducerConsumerTest.java | 9 ++---
.../java/org/apache/pulsar/io/PulsarSinkE2ETest.java| 10 ++
.../pulsar/client/impl/MultiTopicsConsumerImpl.java | 17 +++--
.../apache/pulsar/functions/source/PulsarSource.java| 15 ++-
4 files changed, 25 insertions(+), 26 deletions(-)
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PartitionedProducerConsumerTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PartitionedProducerConsumerTest.java
index a599532..ae7757d 100644
---
a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PartitionedProducerConsumerTest.java
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PartitionedProducerConsumerTest.java
@@ -290,19 +290,14 @@ public class PartitionedProducerConsumerTest extends
ProducerConsumerBase {
}
try {
-producer = pulsarClient.newProducer().topic(topicName.toString())
-.enableBatching(false)
-.messageRoutingMode(MessageRoutingMode.SinglePartition)
-.create();
+producer =
pulsarClient.newProducer().topic(topicName.toString()).enableBatching(false)
+
.messageRoutingMode(MessageRoutingMode.SinglePartition).create();
consumer =
pulsarClient.newConsumer().topic(topicName.toString()).subscriptionName("my-sub").subscribe();
producer.send("message1".getBytes());
producer.send("message2".getBytes());
/* Message msg1 = */ consumer.receive();
Message msg2 = consumer.receive();
consumer.acknowledgeCumulative(msg2);
-Assert.fail("should fail since ack cumulative is not supported for
partitioned topic");
-} catch (PulsarClientException e) {
-Assert.assertTrue(e instanceof
PulsarClientException.NotSupportedException);
} finally {
producer.close();
consumer.unsubscribe();
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarSinkE2ETest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarSinkE2ETest.java
index abc4735..5db9a0a 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarSinkE2ETest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarSinkE2ETest.java
@@ -62,6 +62,7 @@ import org.apache.pulsar.functions.api.utils.IdentityFunction;
import org.apache.pulsar.functions.instance.JavaInstanceRunnable;
import org.apache.pulsar.functions.proto.Function;
import org.apache.pulsar.functions.proto.Function.FunctionDetails;
+import org.apache.pulsar.functions.proto.Function.ProcessingGuarantees;
import org.apache.pulsar.functions.proto.Function.SinkSpec;
import org.apache.pulsar.functions.proto.Function.SourceSpec;
import org.apache.pulsar.functions.proto.InstanceCommunication.FunctionStatus;
@@ -356,11 +357,11 @@ public class PulsarSinkE2ETest {
retryStrategically((test) -> {
try {
SubscriptionStats subStats =
admin.topics().getStats(sourceTopic).subscriptions.get(subscriptionName);
-return subStats.unackedMessages == 0;
+return subStats.unackedMessages == 0 &&
subStats.msgThroughputOut == totalMsgs;
} catch (PulsarAdminException e) {
return false;
}
-}, 5, 500);
+}, 5, 200);
FunctionRuntimeManager functionRuntimeManager =
functionsWorkerService.getFunctionRuntimeManager();
functionRuntimeManager.updateRates();
@@ -399,11 +400,12 @@ public class PulsarSinkE2ETest {
functionDetailsBuilder.setRuntime(FunctionDetails.Runtime.JAVA);
functionDetailsBuilder.setParallelism(1);
functionDetailsBuilder.setClassName(IdentityFunction.class.getName());
+
functionDetailsBuilder.setProce
[GitHub] sijie closed pull request #2347: Fix: function with multi-topic not acking on effectively-once
sijie closed pull request #2347: Fix: function with multi-topic not acking on
effectively-once
URL: https://github.com/apache/incubator-pulsar/pull/2347
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PartitionedProducerConsumerTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PartitionedProducerConsumerTest.java
index a599532bda..ae7757d09e 100644
---
a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PartitionedProducerConsumerTest.java
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PartitionedProducerConsumerTest.java
@@ -290,19 +290,14 @@ public void testSillyUser() throws Exception {
}
try {
-producer = pulsarClient.newProducer().topic(topicName.toString())
-.enableBatching(false)
-.messageRoutingMode(MessageRoutingMode.SinglePartition)
-.create();
+producer =
pulsarClient.newProducer().topic(topicName.toString()).enableBatching(false)
+
.messageRoutingMode(MessageRoutingMode.SinglePartition).create();
consumer =
pulsarClient.newConsumer().topic(topicName.toString()).subscriptionName("my-sub").subscribe();
producer.send("message1".getBytes());
producer.send("message2".getBytes());
/* Message msg1 = */ consumer.receive();
Message msg2 = consumer.receive();
consumer.acknowledgeCumulative(msg2);
-Assert.fail("should fail since ack cumulative is not supported for
partitioned topic");
-} catch (PulsarClientException e) {
-Assert.assertTrue(e instanceof
PulsarClientException.NotSupportedException);
} finally {
producer.close();
consumer.unsubscribe();
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarSinkE2ETest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarSinkE2ETest.java
index abc4735ea1..5db9a0ac04 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarSinkE2ETest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarSinkE2ETest.java
@@ -62,6 +62,7 @@
import org.apache.pulsar.functions.instance.JavaInstanceRunnable;
import org.apache.pulsar.functions.proto.Function;
import org.apache.pulsar.functions.proto.Function.FunctionDetails;
+import org.apache.pulsar.functions.proto.Function.ProcessingGuarantees;
import org.apache.pulsar.functions.proto.Function.SinkSpec;
import org.apache.pulsar.functions.proto.Function.SourceSpec;
import org.apache.pulsar.functions.proto.InstanceCommunication.FunctionStatus;
@@ -356,11 +357,11 @@ public void testPulsarSinkStats() throws Exception {
retryStrategically((test) -> {
try {
SubscriptionStats subStats =
admin.topics().getStats(sourceTopic).subscriptions.get(subscriptionName);
-return subStats.unackedMessages == 0;
+return subStats.unackedMessages == 0 &&
subStats.msgThroughputOut == totalMsgs;
} catch (PulsarAdminException e) {
return false;
}
-}, 5, 500);
+}, 5, 200);
FunctionRuntimeManager functionRuntimeManager =
functionsWorkerService.getFunctionRuntimeManager();
functionRuntimeManager.updateRates();
@@ -399,11 +400,12 @@ protected static FunctionDetails createSinkConfig(String
jarFile, String tenant,
functionDetailsBuilder.setRuntime(FunctionDetails.Runtime.JAVA);
functionDetailsBuilder.setParallelism(1);
functionDetailsBuilder.setClassName(IdentityFunction.class.getName());
+
functionDetailsBuilder.setProcessingGuarantees(ProcessingGuarantees.EFFECTIVELY_ONCE);
// set source spec
// source spec classname should be empty so that the default pulsar
source will be used
SourceSpec.Builder sourceSpecBuilder = SourceSpec.newBuilder();
-
sourceSpecBuilder.setSubscriptionType(Function.SubscriptionType.SHARED);
+
sourceSpecBuilder.setSubscriptionType(Function.SubscriptionType.FAILOVER);
sourceSpecBuilder.setTypeClassName(typeArg.getName());
sourceSpecBuilder.setTopicsPattern(sourceTopicPattern);
sourceSpecBuilder.setSubscriptionName(subscriptionName);
@@ -484,7 +486,7 @@ public void testFileUrlFunctionWithoutPassingTypeArgs()
throws Exception {
// set source spec
// source spec classname should be empty so that the default pulsar
source will be used
SourceSpec.Builder sourceSpecBuilder = SourceSpec.newBuilder();
-
sourceSpecBuilder.setSubscriptionType(Function.SubscriptionType.SHARED);
+
sourceSpecBuilder.s
[GitHub] rdhabalia commented on issue #2354: add function package-url into function-get api response
rdhabalia commented on issue #2354: add function package-url into function-get api response URL: https://github.com/apache/incubator-pulsar/pull/2354#issuecomment-412666891 @sijie @jerrypeng can you please take a look .. we want to know function-pkg url to know actual package-version which function is running. This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] rdhabalia removed a comment on issue #2353: Set correct exception if function runnable fails with Error
rdhabalia removed a comment on issue #2353: Set correct exception if function runnable fails with Error URL: https://github.com/apache/incubator-pulsar/pull/2353#issuecomment-412666125 rerun java8 tests This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] rdhabalia commented on issue #2353: Set correct exception if function runnable fails with Error
rdhabalia commented on issue #2353: Set correct exception if function runnable fails with Error URL: https://github.com/apache/incubator-pulsar/pull/2353#issuecomment-412666125 rerun java8 tests This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] rdhabalia commented on issue #2347: Fix: function with multi-topic not acking on effectively-once
rdhabalia commented on issue #2347: Fix: function with multi-topic not acking on effectively-once URL: https://github.com/apache/incubator-pulsar/pull/2347#issuecomment-412666274 @sijie @merlimat can we merge this PR.? This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] srkukarni commented on a change in pull request #2266: Integrate functions and io with schema registry
srkukarni commented on a change in pull request #2266: Integrate functions and
io with schema registry
URL: https://github.com/apache/incubator-pulsar/pull/2266#discussion_r209760478
##
File path:
pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSources.java
##
@@ -207,10 +201,18 @@ void runCmd() throws Exception {
@Parameter(names = "--processingGuarantees", description = "The
processing guarantees (aka delivery semantics) applied to the Source")
protected FunctionConfig.ProcessingGuarantees processingGuarantees;
-@Parameter(names = "--destinationTopicName", description = "The Pulsar
topic to which data is sent")
+
+@Parameter(names = { "-o", "--destinationTopicName" }, description =
"The Pulsar topic to which data is sent")
protected String destinationTopicName;
-@Parameter(names = "--deserializationClassName", description = "The
SerDe classname for the source")
+
+@Parameter(names = "--deserializationClassName", description = "The
SerDe classname for the source", hidden = true)
protected String deserializationClassName;
Review comment:
shouldn;t this be DEPRECATED_deserialzationClassName
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] rdhabalia opened a new pull request #2365: Add support to restart function
rdhabalia opened a new pull request #2365: Add support to restart function URL: https://github.com/apache/incubator-pulsar/pull/2365 ### Motivation Few times we have seen that our function process is stuck and not consuming messages from pulsar-source or computing as per expectation. So, sometimes it requires to restart stuck function due to issue with function logic or pulsar-framework. Pulsar broker already has similar feature to unload namespace bundles to reinitialize bundle resources. Similar way, restarting function and reinitialize function resources helps in such situation. ### Modifications add REST/CLI to restart function. ### Result Admin can restart function instance. This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] srkukarni commented on a change in pull request #2266: Integrate functions and io with schema registry
srkukarni commented on a change in pull request #2266: Integrate functions and
io with schema registry
URL: https://github.com/apache/incubator-pulsar/pull/2266#discussion_r209757660
##
File path:
pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSinks.java
##
@@ -204,14 +207,27 @@ void runCmd() throws Exception {
@Parameter(names = { "-t", "--sink-type" }, description = "The sinks's
connector provider")
protected String sinkType;
-@Parameter(names = "--inputs", description = "The sink's input topic
or topics (multiple topics can be specified as a comma-separated list)")
+@Parameter(names = { "-i",
+"--inputs" }, description = "The sink's input topic or topics
(multiple topics can be specified as a comma-separated list)")
protected String inputs;
-@Parameter(names = "--topicsPattern", description = "TopicsPattern to
consume from list of topics under a namespace that match the pattern. [--input]
and [--topicsPattern] are mutually exclusive. Add SerDe class name for a
pattern in --customSerdeInputs (supported for java fun only)")
+
+@Parameter(names = "--topicsPattern", description = "TopicsPattern to
consume from list of topics under a namespace that match the pattern. [--input]
and [--topicsPattern] are mutually exclusive. Add SerDe class name for a
pattern in --customSerdeInputs (supported for java fun only)", hidden = true)
protected String topicsPattern;
+
+@Parameter(names = { "-st",
+"--schema-type" }, description = "The builtin schema type (eg:
'avro', 'json', etc..) or the class name for a Schema or Serde implementation")
+protected String schemaTypeOrClassName = "";
+
@Parameter(names = "--subsName", description = "Pulsar source
subscription name if user wants a specific subscription-name for input-topic
consumer")
protected String subsName;
-@Parameter(names = "--customSerdeInputs", description = "The map of
input topics to SerDe class names (as a JSON string)")
+
+@Parameter(names = "--customSerdeInputs", description = "The map of
input topics to SerDe class names (as a JSON string)", hidden = true)
protected String customSerdeInputString;
Review comment:
Shouldn't this be deprecated?
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] srkukarni commented on a change in pull request #2266: Integrate functions and io with schema registry
srkukarni commented on a change in pull request #2266: Integrate functions and
io with schema registry
URL: https://github.com/apache/incubator-pulsar/pull/2266#discussion_r209755250
##
File path:
pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java
##
@@ -226,30 +228,32 @@ void processArguments() throws Exception {
description = "Path to the main Python file for the function
(if the function is written in Python)",
listConverter = StringConverter.class)
protected String pyFile;
-@Parameter(names = "--inputs", description = "The function's input
topic or topics (multiple topics can be specified as a comma-separated list)")
+@Parameter(names = { "-i",
+"--inputs" }, description = "The function's input topic or
topics (multiple topics can be specified as a comma-separated list)")
protected String inputs;
// for backwards compatibility purposes
@Parameter(names = "--topicsPattern", description = "TopicsPattern to
consume from list of topics under a namespace that match the pattern. [--input]
and [--topic-pattern] are mutually exclusive. Add SerDe class name for a
pattern in --custom-serde-inputs (supported for java fun only)", hidden = true)
protected String DEPRECATED_topicsPattern;
@Parameter(names = "--topics-pattern", description = "The topic
pattern to consume from list of topics under a namespace that match the
pattern. [--input] and [--topic-pattern] are mutually exclusive. Add SerDe
class name for a pattern in --custom-serde-inputs (supported for java fun
only)")
protected String topicsPattern;
-@Parameter(names = "--output", description = "The function's output
topic")
+
+@Parameter(names = {"-o", "--output"}, description = "The function's
output topic")
protected String output;
// for backwards compatibility purposes
@Parameter(names = "--logTopic", description = "The topic to which the
function's logs are produced", hidden = true)
protected String DEPRECATED_logTopic;
@Parameter(names = "--log-topic", description = "The topic to which
the function's logs are produced")
protected String logTopic;
+
+@Parameter(names = {"-st", "--schema-type"}, description = "The
builtin schema type or custom schema class name to be used for messages output
by the function")
Review comment:
So does that mean that I cannot consume a avro schema topic and output a
json schema topic?
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] sijie opened a new pull request #2364: [documentation] Cherry-pick tiered storage documentation to new website
sijie opened a new pull request #2364: [documentation] Cherry-pick tiered storage documentation to new website URL: https://github.com/apache/incubator-pulsar/pull/2364 ### Motivation Cherry-pick #2152 This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] sijie commented on issue #2289: Broker suddenly goes down
sijie commented on issue #2289: Broker suddenly goes down URL: https://github.com/apache/incubator-pulsar/issues/2289#issuecomment-412614644 @massakam great finding! so it seems the problems have been addressed in 2.x releases? This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] sijie commented on issue #2195: Can't start Global Zookeeper in a Multi-cluster setup
sijie commented on issue #2195: Can't start Global Zookeeper in a Multi-cluster setup URL: https://github.com/apache/incubator-pulsar/issues/2195#issuecomment-412614078 @dsambandam it seems that there is inconsistent metadata after you copied the metadata. as the log say it has been a smaller zxid. when you copied the metadata, do you first cleanup the global zk directories? This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] sijie commented on a change in pull request #2362: [documentation][deploy] Update deployment instructions for deploying pulsar to minikube
sijie commented on a change in pull request #2362: [documentation][deploy] Update deployment instructions for deploying pulsar to minikube URL: https://github.com/apache/incubator-pulsar/pull/2362#discussion_r209694791 ## File path: deployment/kubernetes/generic/bookie.yaml ## @@ -23,10 +23,10 @@ kind: ConfigMap metadata: name: bookie-config data: -PULSAR_MEM: "\" -Xms512m -Xmx512m -XX:MaxDirectMemorySize=1g\"" -dbStorage_writeCacheMaxSizeMb: "256" # Write cache size (direct memory) -dbStorage_readAheadCacheMaxSizeMb: "256" # Read cache size (direct memory) -zkServers: zk-0.zookeeper,zk-1.zookeeper,zk-2.zookeeper +PULSAR_MEM: "\" -Xms64m -Xmx256m -XX:MaxDirectMemorySize=256m\"" Review comment: the changes are I made here are reducing the memory usage for people to start. so they are still valid for a "custom" K8S cluster, we don't need to differentiate them. because in general, for a "custom" K8S cluster, we don't know exactly the resources people have, so starting from smaller values would be better for most people. This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] merlimat commented on a change in pull request #2362: [documentation][deploy] Update deployment instructions for deploying pulsar to minikube
merlimat commented on a change in pull request #2362: [documentation][deploy] Update deployment instructions for deploying pulsar to minikube URL: https://github.com/apache/incubator-pulsar/pull/2362#discussion_r209669024 ## File path: deployment/kubernetes/generic/bookie.yaml ## @@ -23,10 +23,10 @@ kind: ConfigMap metadata: name: bookie-config data: -PULSAR_MEM: "\" -Xms512m -Xmx512m -XX:MaxDirectMemorySize=1g\"" -dbStorage_writeCacheMaxSizeMb: "256" # Write cache size (direct memory) -dbStorage_readAheadCacheMaxSizeMb: "256" # Read cache size (direct memory) -zkServers: zk-0.zookeeper,zk-1.zookeeper,zk-2.zookeeper +PULSAR_MEM: "\" -Xms64m -Xmx256m -XX:MaxDirectMemorySize=256m\"" Review comment: The "generic" instructions were also meant for a "custom" K8S cluster, though not necessarily a minikube one. Should we rather differentiate the 2 cases? This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] merlimat closed pull request #2361: [docker] Publish pulsar-all image to docker hub
merlimat closed pull request #2361: [docker] Publish pulsar-all image to docker
hub
URL: https://github.com/apache/incubator-pulsar/pull/2361
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/docker/publish.sh b/docker/publish.sh
index 411fd0da31..f73bdacaee 100755
--- a/docker/publish.sh
+++ b/docker/publish.sh
@@ -63,18 +63,22 @@ set -x
set -e
docker tag pulsar:latest ${docker_registry_org}/pulsar:latest
+docker tag pulsar-all:latest ${docker_registry_org}/pulsar-all:latest
docker tag pulsar-grafana:latest ${docker_registry_org}/pulsar-grafana:latest
docker tag pulsar-dashboard:latest
${docker_registry_org}/pulsar-dashboard:latest
docker tag pulsar:latest ${docker_registry_org}/pulsar:$MVN_VERSION
+docker tag pulsar-all:latest ${docker_registry_org}/pulsar-all:$MVN_VERSION
docker tag pulsar-grafana:latest
${docker_registry_org}/pulsar-grafana:$MVN_VERSION
docker tag pulsar-dashboard:latest
${docker_registry_org}/pulsar-dashboard:$MVN_VERSION
# Push all images and tags
docker push ${docker_registry_org}/pulsar:latest
+docker push ${docker_registry_org}/pulsar-all:latest
docker push ${docker_registry_org}/pulsar-grafana:latest
docker push ${docker_registry_org}/pulsar-dashboard:latest
docker push ${docker_registry_org}/pulsar:$MVN_VERSION
+docker push ${docker_registry_org}/pulsar-all:$MVN_VERSION
docker push ${docker_registry_org}/pulsar-grafana:$MVN_VERSION
docker push ${docker_registry_org}/pulsar-dashboard:$MVN_VERSION
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[incubator-pulsar] branch master updated: [docker] Publish pulsar-all image to docker hub (#2361)
This is an automated email from the ASF dual-hosted git repository.
mmerli pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 777223d [docker] Publish pulsar-all image to docker hub (#2361)
777223d is described below
commit 777223d8031233ef03f61c41b8b99916c268b819
Author: Sijie Guo
AuthorDate: Mon Aug 13 09:07:44 2018 -0700
[docker] Publish pulsar-all image to docker hub (#2361)
### Motivation
We have `pulsar-all` image since 2.1.0 release. However this image is not
pushed to docker hub.
### Changes
Publish `pulsar-all` image to docker hub.
---
docker/publish.sh | 4
1 file changed, 4 insertions(+)
diff --git a/docker/publish.sh b/docker/publish.sh
index 411fd0d..f73bdac 100755
--- a/docker/publish.sh
+++ b/docker/publish.sh
@@ -63,18 +63,22 @@ set -x
set -e
docker tag pulsar:latest ${docker_registry_org}/pulsar:latest
+docker tag pulsar-all:latest ${docker_registry_org}/pulsar-all:latest
docker tag pulsar-grafana:latest ${docker_registry_org}/pulsar-grafana:latest
docker tag pulsar-dashboard:latest
${docker_registry_org}/pulsar-dashboard:latest
docker tag pulsar:latest ${docker_registry_org}/pulsar:$MVN_VERSION
+docker tag pulsar-all:latest ${docker_registry_org}/pulsar-all:$MVN_VERSION
docker tag pulsar-grafana:latest
${docker_registry_org}/pulsar-grafana:$MVN_VERSION
docker tag pulsar-dashboard:latest
${docker_registry_org}/pulsar-dashboard:$MVN_VERSION
# Push all images and tags
docker push ${docker_registry_org}/pulsar:latest
+docker push ${docker_registry_org}/pulsar-all:latest
docker push ${docker_registry_org}/pulsar-grafana:latest
docker push ${docker_registry_org}/pulsar-dashboard:latest
docker push ${docker_registry_org}/pulsar:$MVN_VERSION
+docker push ${docker_registry_org}/pulsar-all:$MVN_VERSION
docker push ${docker_registry_org}/pulsar-grafana:$MVN_VERSION
docker push ${docker_registry_org}/pulsar-dashboard:$MVN_VERSION
[GitHub] sijie commented on issue #2363: [documentation][deploy] Improve helm deployment script to deploy Pulsar to minikube
sijie commented on issue #2363: [documentation][deploy] Improve helm deployment script to deploy Pulsar to minikube URL: https://github.com/apache/incubator-pulsar/pull/2363#issuecomment-412465063 for reviewers: this change is based on #2362 , so commit 0c49a8d is the change to review in this PR. This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] sijie opened a new pull request #2363: [documentation][deploy] Improve helm deployment script to deploy Pulsar to minikube
sijie opened a new pull request #2363: [documentation][deploy] Improve helm deployment script to deploy Pulsar to minikube URL: https://github.com/apache/incubator-pulsar/pull/2363 ### Changes - update the helm scripts: bookie/autorecovery/broker pods should wait until metadata is initialized - disable `autoRecovery` on bookies since we start `AutoRecovery` in separate pods - enable function worker on brokers - provide a values file for minikube - update documentation for using helm chart to deploy a cluster to minikube This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[incubator-pulsar] branch asf-site updated: Updated site at revision c0a5072
This is an automated email from the ASF dual-hosted git repository.
mmerli pushed a commit to branch asf-site
in repository https://gitbox.apache.org/repos/asf/incubator-pulsar.git
The following commit(s) were added to refs/heads/asf-site by this push:
new a50747e Updated site at revision c0a5072
a50747e is described below
commit a50747ec201e8731e13490af21be514ccdc3a309
Author: jenkins
AuthorDate: Mon Aug 13 08:35:56 2018 +
Updated site at revision c0a5072
---
content/swagger/swagger.json | 48 ++--
1 file changed, 24 insertions(+), 24 deletions(-)
diff --git a/content/swagger/swagger.json b/content/swagger/swagger.json
index 5a2eb43..c21c847 100644
--- a/content/swagger/swagger.json
+++ b/content/swagger/swagger.json
@@ -6364,10 +6364,10 @@
"type" : "string"
}
},
-"clientVersion" : {
+"connectedSince" : {
"type" : "string"
},
-"connectedSince" : {
+"clientVersion" : {
"type" : "string"
},
"address" : {
@@ -6687,20 +6687,32 @@
"type" : "number",
"format" : "double"
},
-"msgThroughputIn" : {
- "type" : "number",
- "format" : "double"
+"underLoaded" : {
+ "type" : "boolean"
+},
+"overLoaded" : {
+ "type" : "boolean"
+},
+"loadReportType" : {
+ "type" : "string"
},
"msgThroughputOut" : {
"type" : "number",
"format" : "double"
},
+"cpu" : {
+ "$ref" : "#/definitions/ResourceUsage"
+},
+"directMemory" : {
+ "$ref" : "#/definitions/ResourceUsage"
+},
"lastUpdate" : {
"type" : "integer",
"format" : "int64"
},
-"bandwidthIn" : {
- "$ref" : "#/definitions/ResourceUsage"
+"msgThroughputIn" : {
+ "type" : "number",
+ "format" : "double"
},
"bandwidthOut" : {
"$ref" : "#/definitions/ResourceUsage"
@@ -6708,20 +6720,8 @@
"memory" : {
"$ref" : "#/definitions/ResourceUsage"
},
-"cpu" : {
- "$ref" : "#/definitions/ResourceUsage"
-},
-"directMemory" : {
+"bandwidthIn" : {
"$ref" : "#/definitions/ResourceUsage"
-},
-"underLoaded" : {
- "type" : "boolean"
-},
-"overLoaded" : {
- "type" : "boolean"
-},
-"loadReportType" : {
- "type" : "string"
}
}
},
@@ -6881,13 +6881,13 @@
"type" : "number",
"format" : "double"
},
-"clientVersion" : {
+"producerName" : {
"type" : "string"
},
"connectedSince" : {
"type" : "string"
},
-"producerName" : {
+"clientVersion" : {
"type" : "string"
},
"address" : {
@@ -7585,13 +7585,13 @@
"type" : "string"
}
},
-"clientVersion" : {
+"producerName" : {
"type" : "string"
},
"connectedSince" : {
"type" : "string"
},
-"producerName" : {
+"clientVersion" : {
"type" : "string"
},
"address" : {
