cache StorageClient
@@ -47,6 +47,7 @@ import java.util.List;
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 class FunctionsImpl {
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,14 +1077,17 @@ public class FunctionsImpl {
String stateStorageServiceUrl = worker().getWorkerConfig().getStateStorageServiceUrl();
- try (StorageClient client = StorageClientBuilder.newBuilder()
+ 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(client.openTable(tableName))) {
+ .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)
@@ -1106,7 +1112,7 @@ public class FunctionsImpl {
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) {
