This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch backport/24695-to-camel-4.18.x in repository https://gitbox.apache.org/repos/asf/camel.git
commit d61776e5b46ffc65f6dc06bafe88b6f328431e68 Author: Claus Ibsen <[email protected]> AuthorDate: Wed Jul 15 09:26:00 2026 +0200 CAMEL-24061: Wire OAuth credentials onto gRPC stub channel in camel-zeebe (#24695) The managedChannel used by gRPC-stub-based operations (cancelProcess, publishMessage, completeJob, failJob, updateJobRetries, throwError, deployResource) was always built with usePlaintext() and never received OAuth credentials, causing all non-startProcess operations to fail with authentication errors when the Zeebe gateway requires OAuth. Wrap the managedChannel with a ClientInterceptor that applies the OAuthCredentialsProvider to every outgoing gRPC call when credentials are configured. Signed-off-by: Claus Ibsen <[email protected]> Co-authored-by: Claude Opus 4.6 <[email protected]> --- .../component/zeebe/internal/ZeebeService.java | 76 +++++++++++++++++----- 1 file changed, 61 insertions(+), 15 deletions(-) diff --git a/components/camel-zeebe/src/main/java/org/apache/camel/component/zeebe/internal/ZeebeService.java b/components/camel-zeebe/src/main/java/org/apache/camel/component/zeebe/internal/ZeebeService.java index c1e3bb97e1b2..b4571c237d48 100644 --- a/components/camel-zeebe/src/main/java/org/apache/camel/component/zeebe/internal/ZeebeService.java +++ b/components/camel-zeebe/src/main/java/org/apache/camel/component/zeebe/internal/ZeebeService.java @@ -17,6 +17,7 @@ package org.apache.camel.component.zeebe.internal; +import java.io.IOException; import java.time.Duration; import com.fasterxml.jackson.core.JsonProcessingException; @@ -31,8 +32,17 @@ import io.camunda.zeebe.client.impl.oauth.OAuthCredentialsProvider; import io.camunda.zeebe.client.impl.oauth.OAuthCredentialsProviderBuilder; import io.camunda.zeebe.gateway.protocol.GatewayGrpc; import io.camunda.zeebe.gateway.protocol.GatewayOuterClass; +import io.grpc.CallOptions; +import io.grpc.Channel; +import io.grpc.ClientCall; +import io.grpc.ClientInterceptor; +import io.grpc.ClientInterceptors; +import io.grpc.ForwardingClientCall; import io.grpc.ManagedChannel; import io.grpc.ManagedChannelBuilder; +import io.grpc.Metadata; +import io.grpc.MethodDescriptor; +import io.grpc.Status; import io.grpc.StatusRuntimeException; import org.apache.camel.component.zeebe.model.DeploymentRequest; import org.apache.camel.component.zeebe.model.DeploymentResponse; @@ -51,6 +61,7 @@ public class ZeebeService { private ZeebeClient zeebeClient; private ManagedChannel managedChannel; + private Channel grpcChannel; private ObjectMapper objectMapper; @@ -77,15 +88,18 @@ public class ZeebeService { public void doStart() { String gatewayAddress = String.format("%s:%d", gatewayHost, gatewayPort); - if (zeebeClient == null) { - if (clientId != null) { - OAuthCredentialsProvider credentialsProvider = new OAuthCredentialsProviderBuilder() - .authorizationServerUrl(oAuthAPI) - .audience(gatewayAddress) - .clientId(clientId) - .clientSecret(clientSecret) - .build(); + OAuthCredentialsProvider credentialsProvider = null; + if (clientId != null) { + credentialsProvider = new OAuthCredentialsProviderBuilder() + .authorizationServerUrl(oAuthAPI) + .audience(gatewayAddress) + .clientId(clientId) + .clientSecret(clientSecret) + .build(); + } + if (zeebeClient == null) { + if (credentialsProvider != null) { zeebeClient = ZeebeClient.newClientBuilder() .gatewayAddress(gatewayAddress) .credentialsProvider(credentialsProvider) @@ -101,6 +115,9 @@ public class ZeebeService { managedChannel = ManagedChannelBuilder.forAddress(gatewayHost, gatewayPort) .usePlaintext() .build(); + grpcChannel = credentialsProvider != null + ? ClientInterceptors.intercept(managedChannel, new OAuthInterceptor(credentialsProvider)) + : managedChannel; } } @@ -148,7 +165,7 @@ public class ZeebeService { resultMessage.setProcessInstanceKey(processMessage.getProcessInstanceKey()); try { - GatewayGrpc.GatewayBlockingStub stub = GatewayGrpc.newBlockingStub(managedChannel); + GatewayGrpc.GatewayBlockingStub stub = GatewayGrpc.newBlockingStub(grpcChannel); GatewayOuterClass.CancelProcessInstanceResponse cancelProcessInstanceResponse = stub.cancelProcessInstance(GatewayOuterClass.CancelProcessInstanceRequest.newBuilder() .setProcessInstanceKey(processMessage.getProcessInstanceKey()) @@ -170,7 +187,7 @@ public class ZeebeService { resultMessage.setCorrelationKey(message.getCorrelationKey()); try { - GatewayGrpc.GatewayBlockingStub stub = GatewayGrpc.newBlockingStub(managedChannel); + GatewayGrpc.GatewayBlockingStub stub = GatewayGrpc.newBlockingStub(grpcChannel); if (message.getCorrelationKey() == null) { LOG.error("Correlation Key is missing!"); resultMessage.setSuccess(false); @@ -211,7 +228,7 @@ public class ZeebeService { JobResponse resultMessage = new JobResponse(); try { - GatewayGrpc.GatewayBlockingStub stub = GatewayGrpc.newBlockingStub(managedChannel); + GatewayGrpc.GatewayBlockingStub stub = GatewayGrpc.newBlockingStub(grpcChannel); GatewayOuterClass.CompleteJobRequest.Builder builder = GatewayOuterClass.CompleteJobRequest.newBuilder() .setJobKey(message.getJobKey()); if (!message.getVariables().isEmpty()) { @@ -239,7 +256,7 @@ public class ZeebeService { JobResponse resultMessage = new JobResponse(); try { - GatewayGrpc.GatewayBlockingStub stub = GatewayGrpc.newBlockingStub(managedChannel); + GatewayGrpc.GatewayBlockingStub stub = GatewayGrpc.newBlockingStub(grpcChannel); GatewayOuterClass.FailJobRequest.Builder builder = GatewayOuterClass.FailJobRequest.newBuilder() .setJobKey(message.getJobKey()); builder = builder.setRetries(message.getRetries()); @@ -262,7 +279,7 @@ public class ZeebeService { JobResponse resultMessage = new JobResponse(); try { - GatewayGrpc.GatewayBlockingStub stub = GatewayGrpc.newBlockingStub(managedChannel); + GatewayGrpc.GatewayBlockingStub stub = GatewayGrpc.newBlockingStub(grpcChannel); GatewayOuterClass.UpdateJobRetriesRequest.Builder builder = GatewayOuterClass.UpdateJobRetriesRequest.newBuilder() .setJobKey(message.getJobKey()); builder = builder.setRetries(message.getRetries()); @@ -284,7 +301,7 @@ public class ZeebeService { JobResponse resultMessage = new JobResponse(); try { - GatewayGrpc.GatewayBlockingStub stub = GatewayGrpc.newBlockingStub(managedChannel); + GatewayGrpc.GatewayBlockingStub stub = GatewayGrpc.newBlockingStub(grpcChannel); GatewayOuterClass.ThrowErrorRequest.Builder builder = GatewayOuterClass.ThrowErrorRequest.newBuilder() .setJobKey(message.getJobKey()); builder = builder.setErrorMessage(message.getErrorMessage()); @@ -307,7 +324,7 @@ public class ZeebeService { DeploymentResponse resultMessage = new DeploymentResponse(); try { - GatewayGrpc.GatewayBlockingStub stub = GatewayGrpc.newBlockingStub(managedChannel); + GatewayGrpc.GatewayBlockingStub stub = GatewayGrpc.newBlockingStub(grpcChannel); GatewayOuterClass.Resource resource = GatewayOuterClass.Resource.newBuilder() .setName(message.getName()) .setContent(ByteString.copyFrom(message.getContent())) @@ -360,4 +377,33 @@ public class ZeebeService { public JobWorker registerJobHandler(JobHandler handler, String jobType, int timeout) { return zeebeClient.newWorker().jobType(jobType).handler(handler).timeout(Duration.ofSeconds(timeout)).open(); } + + private static class OAuthInterceptor implements ClientInterceptor { + + private final OAuthCredentialsProvider credentialsProvider; + + OAuthInterceptor(OAuthCredentialsProvider credentialsProvider) { + this.credentialsProvider = credentialsProvider; + } + + @Override + public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall( + MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) { + return new ForwardingClientCall.SimpleForwardingClientCall<>(next.newCall(method, callOptions)) { + @Override + public void start(Listener<RespT> responseListener, Metadata headers) { + try { + credentialsProvider.applyCredentials((key, value) -> headers.put( + Metadata.Key.of(key, Metadata.ASCII_STRING_MARSHALLER), value)); + } catch (IOException e) { + throw Status.UNAUTHENTICATED + .withDescription("Failed to apply OAuth credentials to gRPC call") + .withCause(e) + .asRuntimeException(); + } + super.start(responseListener, headers); + } + }; + } + } }
