From eb30428472b9ba885ee864d30e14c1ee9873d4c3 Mon Sep 17 00:00:00 2001 From: fjtirado Date: Mon, 29 Jun 2026 14:44:56 +0200 Subject: [PATCH] [Fix #1496] Closing managed channel Signed-off-by: fjtirado --- .../executors/grpc/GrpcChannelResolver.java | 42 ------------------ .../impl/executors/grpc/GrpcExecutor.java | 43 +++++++++++++------ 2 files changed, 30 insertions(+), 55 deletions(-) delete mode 100644 impl/grpc/src/main/java/io/serverlessworkflow/impl/executors/grpc/GrpcChannelResolver.java diff --git a/impl/grpc/src/main/java/io/serverlessworkflow/impl/executors/grpc/GrpcChannelResolver.java b/impl/grpc/src/main/java/io/serverlessworkflow/impl/executors/grpc/GrpcChannelResolver.java deleted file mode 100644 index dabde19be..000000000 --- a/impl/grpc/src/main/java/io/serverlessworkflow/impl/executors/grpc/GrpcChannelResolver.java +++ /dev/null @@ -1,42 +0,0 @@ -/* - * Copyright 2020-Present The Serverless Workflow Specification Authors - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package io.serverlessworkflow.impl.executors.grpc; - -import io.grpc.Channel; -import io.grpc.ManagedChannelBuilder; -import io.serverlessworkflow.impl.TaskContext; -import io.serverlessworkflow.impl.WorkflowContext; - -class GrpcChannelResolver { - - static final String GRPC_CHANNEL_PROVIDER = "grpcChannelProvider"; - - static Channel channel( - WorkflowContext workflowContext, - TaskContext taskContext, - GrpcRequestContext grpcRequestContext) { - return workflowContext - .definition() - .application() - .additionalObject(GRPC_CHANNEL_PROVIDER, workflowContext, taskContext) - .orElseGet( - () -> - ManagedChannelBuilder.forAddress( - grpcRequestContext.address(), grpcRequestContext.port()) - .usePlaintext() - .build()); - } -} diff --git a/impl/grpc/src/main/java/io/serverlessworkflow/impl/executors/grpc/GrpcExecutor.java b/impl/grpc/src/main/java/io/serverlessworkflow/impl/executors/grpc/GrpcExecutor.java index 08fa2de20..930e1bc1c 100644 --- a/impl/grpc/src/main/java/io/serverlessworkflow/impl/executors/grpc/GrpcExecutor.java +++ b/impl/grpc/src/main/java/io/serverlessworkflow/impl/executors/grpc/GrpcExecutor.java @@ -23,6 +23,8 @@ import io.grpc.CallOptions; import io.grpc.Channel; import io.grpc.ClientCall; +import io.grpc.ManagedChannel; +import io.grpc.ManagedChannelBuilder; import io.grpc.MethodDescriptor; import io.grpc.protobuf.ProtoUtils; import io.grpc.stub.ClientCalls; @@ -36,10 +38,13 @@ import io.serverlessworkflow.impl.executors.CallableTask; import java.util.Map; import java.util.Objects; +import java.util.Optional; import java.util.concurrent.CompletableFuture; public class GrpcExecutor implements CallableTask { + public static final String GRPC_CHANNEL_PROVIDER = "grpcChannelProvider"; + private final GrpcRequestContext requestContext; private final WorkflowValueResolver> arguments; private final FileDescriptorProto fileDescriptorProto; @@ -63,8 +68,6 @@ public CompletableFuture apply( private CompletableFuture buildGrpcCallExecutor( WorkflowContext workflowContext, TaskContext taskContext, Map arguments) { - Channel channel = GrpcChannelResolver.channel(workflowContext, taskContext, requestContext); - try { Descriptors.FileDescriptor fileDescriptor = Descriptors.FileDescriptor.buildFrom( @@ -82,19 +85,33 @@ private CompletableFuture buildGrpcCallExecutor( MethodDescriptor.MethodType methodType = ProtobufMessageUtils.getMethodType(methodDescriptor); + Optional providedChannel = + workflowContext + .definition() + .application() + .additionalObject(GRPC_CHANNEL_PROVIDER, workflowContext, taskContext); + Channel channel = + providedChannel.orElseGet( + () -> + ManagedChannelBuilder.forAddress(requestContext.address(), requestContext.port()) + .usePlaintext() + .build()); ClientCall call = buildClientCall(channel, methodType, serviceDescriptor, methodDescriptor); - - return switch (methodType) { - case CLIENT_STREAMING -> - handleClientStreaming(workflowContext, arguments, methodDescriptor, call); - case BIDI_STREAMING -> - handleBidiStreaming(workflowContext, arguments, methodDescriptor, call); - case SERVER_STREAMING -> - handleServerStreaming(workflowContext, methodDescriptor, arguments, call); - case UNARY, UNKNOWN -> handleAsyncUnary(workflowContext, methodDescriptor, arguments, call); - }; - + CompletableFuture result = + switch (methodType) { + case CLIENT_STREAMING -> + handleClientStreaming(workflowContext, arguments, methodDescriptor, call); + case BIDI_STREAMING -> + handleBidiStreaming(workflowContext, arguments, methodDescriptor, call); + case SERVER_STREAMING -> + handleServerStreaming(workflowContext, methodDescriptor, arguments, call); + case UNARY, UNKNOWN -> + handleAsyncUnary(workflowContext, methodDescriptor, arguments, call); + }; + return providedChannel.isEmpty() && channel instanceof ManagedChannel managedChannel + ? result.whenComplete((__, ___) -> managedChannel.shutdown()) + : result; } catch (Descriptors.DescriptorValidationException | InvalidProtocolBufferException e) { throw new WorkflowException(WorkflowError.runtime(taskContext, e).build()); }