This is an automated email from the ASF dual-hosted git repository.
jerrypeng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 0851c75 Added api for putting state via cli/rest (#4311)
0851c75 is described below
commit 0851c75a5b1f3bda4fd7943bbd568888079c9383
Author: Sanjeev Kulkarni <sanjee...@gmail.com>
AuthorDate: Mon Jun 3 17:39:48 2019 -0700
Added api for putting state via cli/rest (#4311)
* Added api for putting state via cli/rest
* Addressed comments
* More comments
* Took feedback into account
---
.../pulsar/broker/admin/impl/FunctionsBase.java | 20 +++++
.../functions/worker/PulsarFunctionStateTest.java | 13 +++
.../org/apache/pulsar/client/admin/Functions.java | 27 ++++++
.../client/admin/internal/FunctionsImpl.java | 17 ++++
.../org/apache/pulsar/admin/cli/CmdFunctions.java | 28 ++++++-
.../pulsar/common/functions/FunctionState.java | 1 +
.../python-examples/wordcount_function.py | 1 +
.../functions/worker/rest/api/ComponentImpl.java | 98 ++++++++++++++++++++--
.../worker/rest/api/v3/FunctionsApiV3Resource.java | 11 +++
9 files changed, 208 insertions(+), 8 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 21ecd2e..ec94d04 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
@@ -277,6 +277,26 @@ public class FunctionsBase extends AdminResource
implements Supplier<WorkerServi
}
@POST
+ @ApiOperation(
+ value = "Put the state associated with a Pulsar Function"
+ )
+ @ApiResponses(value = {
+ @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 403, message = "The requester doesn't have
admin permissions"),
+ @ApiResponse(code = 404, message = "The function does not exist"),
+ @ApiResponse(code = 500, message = "Internal server error")
+ })
+ @Path("/{tenant}/{namespace}/{functionName}/state/{key}")
+ @Consumes(MediaType.MULTIPART_FORM_DATA)
+ public void putFunctionState(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String
namespace,
+ final @PathParam("functionName") String
functionName,
+ final @PathParam("key") String key,
+ final @FormDataParam("state") FunctionState
stateJson) {
+ functions.putFunctionState(tenant, namespace, functionName, key,
stateJson, clientAppId(), clientAuthData());
+ }
+
+ @POST
@ApiOperation(value = "Restart function instance", response = Void.class)
@ApiResponses(value = {
@ApiResponse(code = 400, message = "Invalid request"),
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionStateTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionStateTest.java
index 70f5cf7..0b5c340 100644
---
a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionStateTest.java
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionStateTest.java
@@ -377,6 +377,19 @@ public class PulsarFunctionStateTest {
Assert.assertEquals(e.getStatusCode(),
Response.Status.NOT_FOUND.getStatusCode());
}
+ FunctionState newState = new FunctionState("foobar", "foobarvalue",
null, 0l, 0l);
+ try {
+ admin.functions().putFunctionState(tenant, namespacePortion,
functionName + "bar", newState);
+ Assert.fail("Should have failed since function doesn't exist");
+ } catch (PulsarAdminException e) {
+ Assert.assertEquals(e.getStatusCode(),
Response.Status.NOT_FOUND.getStatusCode());
+ }
+
+ // This succeeds because function name is correct
+ admin.functions().putFunctionState(tenant, namespacePortion,
functionName, newState);
+ state = admin.functions().getFunctionState(tenant, namespacePortion,
functionName, "foobar");
+ Assert.assertTrue(state.getStringValue().equals("foobarvalue"));
+
// validate pulsar-sink consumer has consumed all messages and
delivered to Pulsar sink but unacked messages
// due to publish failure
assertNotEquals(admin.topics().getStats(sourceTopic).subscriptions.values().iterator().next().unackedMessages,
diff --git
a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Functions.java
b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Functions.java
index a3edb8a..977d195 100644
---
a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Functions.java
+++
b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Functions.java
@@ -485,4 +485,31 @@ public interface Functions {
* Unexpected error
*/
FunctionState getFunctionState(String tenant, String namespace, String
function, String key) throws PulsarAdminException;
+
+ /**
+ * Puts the given state associated with a Pulsar Function.
+ * <p>
+ * Response Example:
+ *
+ * <pre>
+ * <code>{ "value : 12, version : 2"}</code>
+ * </pre>
+ *
+ * @param tenant
+ * Tenant name
+ * @param namespace
+ * Namespace name
+ * @param function
+ * Function name
+ * @param state
+ * FunctionState
+ **
+ * @throws NotAuthorizedException
+ * You don't have admin permission to get the configuration of
the cluster
+ * @throws NotFoundException
+ * Cluster doesn't exist
+ * @throws PulsarAdminException
+ * Unexpected error
+ */
+ void putFunctionState(String tenant, String namespace, String function,
FunctionState state) throws PulsarAdminException;
}
diff --git
a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/FunctionsImpl.java
b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/FunctionsImpl.java
index 2cb916d..28cfabb 100644
---
a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/FunctionsImpl.java
+++
b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/FunctionsImpl.java
@@ -463,6 +463,7 @@ public class FunctionsImpl extends ComponentResource
implements Functions {
}
}
+ @Override
public FunctionState getFunctionState(String tenant, String namespace,
String function, String key)
throws PulsarAdminException {
try {
@@ -476,4 +477,20 @@ public class FunctionsImpl extends ComponentResource
implements Functions {
throw getApiException(e);
}
}
+
+ @Override
+ public void putFunctionState(String tenant, String namespace, String
function, FunctionState state)
+ throws PulsarAdminException {
+ try {
+ RequestBuilder builder =
post(functions.path(tenant).path(namespace).path(function).path("state").path(state.getKey()).getUri().toASCIIString());
+ builder.addBodyPart(new StringPart("state",
ObjectMapperFactory.getThreadLocal().writeValueAsString(state),
MediaType.APPLICATION_JSON));
+ org.asynchttpclient.Response response =
asyncHttpClient.executeRequest(addAuthHeaders(functions,
builder).build()).get();
+
+ if (response.getStatusCode() < 200 || response.getStatusCode() >=
300) {
+ throw
getApiException(Response.status(response.getStatusCode()).entity(response.getResponseBody()).build());
+ }
+ } catch (Exception e) {
+ throw getApiException(e);
+ }
+ }
}
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 eab62d8..deec5ff 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
@@ -27,6 +27,7 @@ import com.beust.jcommander.Parameter;
import com.beust.jcommander.ParameterException;
import com.beust.jcommander.Parameters;
import com.beust.jcommander.converters.StringConverter;
+import com.fasterxml.jackson.core.type.TypeReference;
import com.google.common.annotations.VisibleForTesting;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
@@ -54,6 +55,7 @@ import org.apache.pulsar.common.functions.UpdateOptions;
import org.apache.pulsar.common.functions.Utils;
import org.apache.pulsar.common.functions.WindowConfig;
import org.apache.pulsar.common.functions.FunctionState;
+import org.apache.pulsar.common.util.ObjectMapperFactory;
@Slf4j
@Parameters(commandDescription = "Interface for managing Pulsar Functions
(lightweight, Lambda-style compute processes that work with Pulsar)")
@@ -71,6 +73,7 @@ public class CmdFunctions extends CmdBase {
private final StartFunction start;
private final ListFunctions lister;
private final StateGetter stateGetter;
+ private final StatePutter statePutter;
private final TriggerFunction triggerer;
private final UploadFunction uploader;
private final DownloadFunction downloader;
@@ -799,7 +802,7 @@ public class CmdFunctions extends CmdBase {
}
}
- @Parameters(commandDescription = "Fetch the current state associated with
a Pulsar Function running in cluster mode")
+ @Parameters(commandDescription = "Fetch the current state associated with
a Pulsar Function")
class StateGetter extends FunctionCommand {
@Parameter(names = { "-k", "--key" }, description = "key")
@@ -830,6 +833,22 @@ public class CmdFunctions extends CmdBase {
}
}
+ @Parameters(commandDescription = "Put the state associated with a Pulsar
Function")
+ class StatePutter extends FunctionCommand {
+
+ @Parameter(names = { "-s", "--state" }, description = "The
FunctionState that needs to be put", required = true)
+ private String state = null;
+
+ @Override
+ void runCmd() throws Exception {
+ TypeReference<FunctionState> typeRef
+ = new TypeReference<FunctionState>() {};
+ FunctionState stateRepr =
ObjectMapperFactory.getThreadLocal().readValue(state, typeRef);
+ admin.functions()
+ .putFunctionState(tenant, namespace, functionName,
stateRepr);
+ }
+ }
+
@Parameters(commandDescription = "Triggers the specified Pulsar Function
with a supplied value")
class TriggerFunction extends FunctionCommand {
// for backward compatibility purposes
@@ -943,6 +962,7 @@ public class CmdFunctions extends CmdBase {
functionStats = new GetFunctionStats();
lister = new ListFunctions();
stateGetter = new StateGetter();
+ statePutter = new StatePutter();
triggerer = new TriggerFunction();
uploader = new UploadFunction();
downloader = new DownloadFunction();
@@ -962,6 +982,7 @@ public class CmdFunctions extends CmdBase {
jcommander.addCommand("stats", getFunctionStats());
jcommander.addCommand("list", getLister());
jcommander.addCommand("querystate", getStateGetter());
+ jcommander.addCommand("putstate", getStatePutter());
jcommander.addCommand("trigger", getTriggerer());
jcommander.addCommand("upload", getUploader());
jcommander.addCommand("download", getDownloader());
@@ -1001,6 +1022,11 @@ public class CmdFunctions extends CmdBase {
}
@VisibleForTesting
+ StatePutter getStatePutter() {
+ return statePutter;
+ }
+
+ @VisibleForTesting
StateGetter getStateGetter() {
return stateGetter;
}
diff --git
a/pulsar-common/src/main/java/org/apache/pulsar/common/functions/FunctionState.java
b/pulsar-common/src/main/java/org/apache/pulsar/common/functions/FunctionState.java
index 15bcda0..5b7b0ca 100644
---
a/pulsar-common/src/main/java/org/apache/pulsar/common/functions/FunctionState.java
+++
b/pulsar-common/src/main/java/org/apache/pulsar/common/functions/FunctionState.java
@@ -34,6 +34,7 @@ import lombok.*;
public class FunctionState {
private String key;
private String stringValue;
+ private byte[] byteValue;
private Long numberValue;
private Long version;
}
diff --git a/pulsar-functions/python-examples/wordcount_function.py
b/pulsar-functions/python-examples/wordcount_function.py
index d84eb95..22cf4d7 100644
--- a/pulsar-functions/python-examples/wordcount_function.py
+++ b/pulsar-functions/python-examples/wordcount_function.py
@@ -31,4 +31,5 @@ class WordCountFunction(Function):
words = input.split()
for word in words:
context.incr_counter(word, 1)
+ context.get_logger().info("The value is " +
str(context.get_counter(word)))
return input + "!"
diff --git
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/ComponentImpl.java
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/ComponentImpl.java
index 2dba1cd..62ac9bd 100644
---
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/ComponentImpl.java
+++
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/ComponentImpl.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.functions.worker.rest.api;
+import com.fasterxml.jackson.core.type.TypeReference;
import com.google.gson.Gson;
import com.google.protobuf.ByteString;
import io.netty.buffer.ByteBuf;
@@ -56,6 +57,7 @@ import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.FunctionStats;
import org.apache.pulsar.common.policies.data.TenantInfo;
import org.apache.pulsar.common.util.Codec;
+import org.apache.pulsar.common.util.ObjectMapperFactory;
import org.apache.pulsar.functions.auth.FunctionAuthData;
import org.apache.pulsar.functions.instance.InstanceUtils;
import org.apache.pulsar.functions.proto.Function;
@@ -1441,7 +1443,7 @@ public abstract class ComponentImpl {
// validate parameters
try {
- validateGetFunctionStateParams(tenant, namespace, functionName,
key);
+ validateFunctionStateParams(tenant, namespace, functionName, key);
} catch (IllegalArgumentException e) {
log.error("Invalid getFunctionState request @ /{}/{}/{}/{}",
tenant, namespace, functionName, key, e);
@@ -1470,9 +1472,13 @@ public abstract class ComponentImpl {
throw new RestException(Status.NOT_FOUND, "key '" + key +
"' doesn't exist.");
} else {
if (kv.isNumber()) {
- value = new FunctionState(key, null, kv.numberValue(),
kv.version());
+ value = new FunctionState(key, null, null,
kv.numberValue(), kv.version());
} else {
- value = new FunctionState(key, new
String(ByteBufUtil.getBytes(kv.value()), UTF_8), null, kv.version());
+ try {
+ value = new FunctionState(key, new
String(ByteBufUtil.getBytes(kv.value(), kv.value().readerIndex(),
kv.value().readableBytes()), UTF_8), null, null, kv.version());
+ } catch (Exception e) {
+ value = new FunctionState(key, null,
ByteBufUtil.getBytes(kv.value()), null, kv.version());
+ }
}
}
}
@@ -1490,6 +1496,84 @@ public abstract class ComponentImpl {
return value;
}
+ public void putFunctionState(final String tenant,
+ final String namespace,
+ final String functionName,
+ final String key,
+ final FunctionState state,
+ final String clientRole,
+ final AuthenticationDataSource
clientAuthenticationDataHttps) {
+
+ if (!isWorkerServiceAvailable()) {
+ throwUnavailableException();
+ }
+
+ if (null == worker().getStateStoreAdminClient()) {
+ throwStateStoreUnvailableResponse();
+ }
+
+ try {
+ if (!isAuthorizedRole(tenant, namespace, clientRole,
clientAuthenticationDataHttps)) {
+ log.error("{}/{}/{} Client [{}] is not admin and authorized to
put state for {}", tenant, namespace,
+ functionName, clientRole,
ComponentTypeUtils.toString(componentType));
+ throw new RestException(Status.UNAUTHORIZED, "client is not
authorize to perform operation");
+ }
+ } catch (PulsarAdminException e) {
+ log.error("{}/{}/{} Failed to authorize [{}]", tenant, namespace,
functionName, e);
+ throw new RestException(Status.INTERNAL_SERVER_ERROR,
e.getMessage());
+ }
+
+ if (!key.equals(state.getKey())) {
+ log.error("{}/{}/{} Bad putFunction Request, path key doesn't
match key in json", tenant, namespace, functionName);
+ throw new RestException(Status.BAD_REQUEST, "Path key doesn't
match key in json");
+ }
+ if (state.getStringValue() == null && state.getByteValue() == null) {
+ throw new RestException(Status.BAD_REQUEST, "Setting Counter
values not supported in put state");
+ }
+
+ // validate parameters
+ try {
+ validateFunctionStateParams(tenant, namespace, functionName, key);
+ } catch (IllegalArgumentException e) {
+ log.error("Invalid putFunctionState request @ /{}/{}/{}/{}",
+ tenant, namespace, functionName, key, e);
+ throw new RestException(Status.BAD_REQUEST, e.getMessage());
+ }
+
+ String tableNs = getStateNamespace(tenant, namespace);
+ String tableName = functionName;
+
+ String stateStorageServiceUrl =
worker().getWorkerConfig().getStateStorageServiceUrl();
+
+ if (storageClient.get() == null) {
+ storageClient.compareAndSet(null, StorageClientBuilder.newBuilder()
+ .withSettings(StorageClientSettings.newBuilder()
+ .serviceUri(stateStorageServiceUrl)
+ .clientName("functions-admin")
+ .build())
+ .withNamespace(tableNs)
+ .build());
+ }
+
+ ByteBuf value;
+ if (!isEmpty(state.getStringValue())) {
+ value = Unpooled.wrappedBuffer(state.getStringValue().getBytes());
+ } else {
+ value = Unpooled.wrappedBuffer(state.getByteValue());
+ }
+ try (Table<ByteBuf, ByteBuf> table =
result(storageClient.get().openTable(tableName))) {
+ result(table.put(Unpooled.wrappedBuffer(key.getBytes(UTF_8)),
value));
+ } catch
(org.apache.bookkeeper.clients.exceptions.NamespaceNotFoundException |
org.apache.bookkeeper.clients.exceptions.StreamNotFoundException e) {
+ log.error("Error while putFunctionState request @ /{}/{}/{}/{}",
+ tenant, namespace, functionName, key, e);
+ throw new RestException(Status.NOT_FOUND, e.getMessage());
+ } catch (Exception e) {
+ log.error("Error while putFunctionState request @ /{}/{}/{}/{}",
+ tenant, namespace, functionName, key, e);
+ throw new RestException(Status.INTERNAL_SERVER_ERROR,
e.getMessage());
+ }
+ }
+
public void uploadFunction(final InputStream uploadedInputStream, final
String path) {
// validate parameters
try {
@@ -1586,10 +1670,10 @@ public abstract class ComponentImpl {
}
}
- private void validateGetFunctionStateParams(final String tenant,
- final String namespace,
- final String functionName,
- final String key)
+ private void validateFunctionStateParams(final String tenant,
+ final String namespace,
+ final String functionName,
+ final String key)
throws IllegalArgumentException {
if (tenant == null) {
diff --git
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v3/FunctionsApiV3Resource.java
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v3/FunctionsApiV3Resource.java
index 38f4c4b..2f1a87f 100644
---
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v3/FunctionsApiV3Resource.java
+++
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v3/FunctionsApiV3Resource.java
@@ -324,4 +324,15 @@ public class FunctionsApiV3Resource extends
FunctionApiResource {
final @PathParam("key") String key)
throws IOException {
return functions.getFunctionState(tenant, namespace, functionName,
key, clientAppId(), clientAuthData());
}
+
+ @POST
+ @Path("/{tenant}/{namespace}/{functionName}/state/{key}")
+ @Consumes(MediaType.MULTIPART_FORM_DATA)
+ public void putFunctionState(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String
namespace,
+ final @PathParam("functionName") String
functionName,
+ final @PathParam("key") String key,
+ final @FormDataParam("state") FunctionState
stateJson) throws IOException {
+ functions.putFunctionState(tenant, namespace, functionName, key,
stateJson, clientAppId(), clientAuthData());
+ }
}
