You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by pe...@apache.org on 2021/02/10 00:15:52 UTC
[pulsar] branch master updated: [Functions] Call the corresponding
restart according to the componenttype. #9502 (#9519)
This is an automated email from the ASF dual-hosted git repository.
penghui 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 1f1b96d [Functions] Call the corresponding restart according to the componenttype. #9502 (#9519)
1f1b96d is described below
commit 1f1b96dd7fe79937733c4e796217fc5eb6359e31
Author: hexiaocheng <he...@163.com>
AuthorDate: Wed Feb 10 08:15:24 2021 +0800
[Functions] Call the corresponding restart according to the componenttype. #9502 (#9519)
Fixes #9502
Add judgment on component type，Call the corresponding restart according to the componenttype
### Motivation
Restart all instances of a Pulsar Source failed.
When I call the "https://pulsar.apache.org/admin/v3/sources/{tenant}/{namespace}/{sourceName}/restart"
See error "Failed to perform http post request: javax.ws.rs.InternalServerErrorException: HTTP 500 Internal Server Error
Function xxxx doesn't exist".
The reason for the problem is that the corresponding functionadmin is not found according to the componenttype, which leads to the restart of sink or source using the functionadmin of function
---
.../functions/worker/FunctionRuntimeManager.java | 27 +++++++++++++++++++---
1 file changed, 24 insertions(+), 3 deletions(-)
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionRuntimeManager.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionRuntimeManager.java
index c95475a..045d43a 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionRuntimeManager.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionRuntimeManager.java
@@ -57,6 +57,7 @@ import org.apache.pulsar.common.util.Reflections;
import org.apache.pulsar.functions.auth.FunctionAuthProvider;
import org.apache.pulsar.functions.proto.Function;
import org.apache.pulsar.functions.proto.Function.Assignment;
+import org.apache.pulsar.functions.proto.Function.FunctionDetails.ComponentType;
import org.apache.pulsar.functions.runtime.RuntimeCustomizer;
import org.apache.pulsar.functions.runtime.RuntimeFactory;
import org.apache.pulsar.functions.runtime.RuntimeSpawner;
@@ -419,7 +420,16 @@ public class FunctionRuntimeManager implements AutoCloseable{
.type(MediaType.APPLICATION_JSON)
.entity(new ErrorData(fullFunctionName + " has not been assigned yet")).build());
}
- this.functionAdmin.functions().restartFunction(tenant, namespace, functionName);
+
+ ComponentType componentType = assignment.getInstance().getFunctionMetaData().getFunctionDetails().getComponentType();
+
+ if (ComponentType.SOURCE == componentType) {
+ this.functionAdmin.sources().restartSource(tenant, namespace, functionName);
+ } else if (ComponentType.SINK == componentType) {
+ this.functionAdmin.sinks().restartSink(tenant, namespace, functionName);
+ } else {
+ this.functionAdmin.functions().restartFunction(tenant, namespace, functionName);
+ }
}
} else {
for (Assignment assignment : assignments) {
@@ -442,8 +452,19 @@ public class FunctionRuntimeManager implements AutoCloseable{
}
continue;
}
- this.functionAdmin.functions().restartFunction(tenant, namespace, functionName,
- assignment.getInstance().getInstanceId());
+
+ ComponentType componentType = assignment.getInstance().getFunctionMetaData().getFunctionDetails().getComponentType();
+
+ if (ComponentType.SOURCE == componentType) {
+ this.functionAdmin.sources().restartSource(tenant, namespace, functionName,
+ assignment.getInstance().getInstanceId());
+ } else if (ComponentType.SINK == componentType) {
+ this.functionAdmin.sinks().restartSink(tenant, namespace, functionName,
+ assignment.getInstance().getInstanceId());
+ } else {
+ this.functionAdmin.functions().restartFunction(tenant, namespace, functionName,
+ assignment.getInstance().getInstanceId());
+ }
}
}
}