srkukarni closed pull request #3078: Issue #2658: cache StorageClient to avoid
create it each time.
URL: https://github.com/apache/pulsar/pull/3078
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/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 474032e0a6..7f3ed44c07 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
@@ -47,6 +47,7 @@
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Supplier;
import javax.ws.rs.WebApplicationException;
@@ -121,6 +122,8 @@
public static final String SOURCE = "Source";
public static final String SINK = "Sink";
+ private final AtomicReference<StorageClient> storageClient = new
AtomicReference<>();
+
private final Supplier<WorkerService> workerServiceSupplier;
public FunctionsImpl(Supplier<WorkerService> workerServiceSupplier) {
@@ -1074,39 +1077,42 @@ public Response getFunctionState(final String tenant,
final String namespace,
String stateStorageServiceUrl =
worker().getWorkerConfig().getStateStorageServiceUrl();
- try (StorageClient client = StorageClientBuilder.newBuilder()
- .withSettings(StorageClientSettings.newBuilder()
- .serviceUri(stateStorageServiceUrl)
- .clientName("functions-admin")
- .build())
- .withNamespace(tableNs)
- .build()) {
- try (Table<ByteBuf, ByteBuf> table =
result(client.openTable(tableName))) {
- try (KeyValue<ByteBuf, ByteBuf> kv =
result(table.getKv(Unpooled.wrappedBuffer(key.getBytes(UTF_8))))) {
- if (null == kv) {
- return Response.status(Status.NOT_FOUND)
- .entity(new String("key '" + key + "' doesn't
exist."))
- .build();
+ if (storageClient.get() == null) {
+ storageClient.compareAndSet(null, StorageClientBuilder.newBuilder()
+ .withSettings(StorageClientSettings.newBuilder()
+ .serviceUri(stateStorageServiceUrl)
+ .clientName("functions-admin")
+ .build())
+ .withNamespace(tableNs)
+ .build());
+ }
+
+ try (Table<ByteBuf, ByteBuf> table =
result(storageClient.get().openTable(tableName))) {
+ try (KeyValue<ByteBuf, ByteBuf> kv =
result(table.getKv(Unpooled.wrappedBuffer(key.getBytes(UTF_8))))) {
+ if (null == kv) {
+ return Response.status(Status.NOT_FOUND)
+ .entity(new String("key '" + key + "' doesn't exist."))
+ .build();
+ } else {
+ String value;
+ if (kv.isNumber()) {
+ value = "value : " + kv.numberValue() + ", version : "
+ kv.version();
} else {
- String value;
- if (kv.isNumber()) {
- value = "value : " + kv.numberValue() + ", version
: " + kv.version();
- } else {
- value = "value : " + new
String(ByteBufUtil.getBytes(kv.value()), UTF_8)
- + ", version : " + kv.version();
- }
- return Response.status(Status.OK)
- .entity(new String(value))
- .build();
+ value = "value : " + new
String(ByteBufUtil.getBytes(kv.value()), UTF_8)
+ + ", version : " + kv.version();
}
+ return Response.status(Status.OK)
+ .entity(new String(value))
+ .build();
}
- } catch (Exception e) {
- log.error("Error while getFunctionState request @
/{}/{}/{}/{}",
- tenant, namespace, functionName, key, e);
- return
Response.status(Status.INTERNAL_SERVER_ERROR).type(MediaType.APPLICATION_JSON)
- .entity(new ErrorData(e.getMessage())).build();
}
+ } catch (Exception e) {
+ log.error("Error while getFunctionState request @ /{}/{}/{}/{}",
+ tenant, namespace, functionName, key, e);
+ return
Response.status(Status.INTERNAL_SERVER_ERROR).type(MediaType.APPLICATION_JSON)
+ .entity(new ErrorData(e.getMessage())).build();
}
+
}
public Response uploadFunction(final InputStream uploadedInputStream,
final String path) {
----------------------------------------------------------------
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:
[email protected]
With regards,
Apache Git Services
