This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 89f6a09c83da CAMEL-24061: Wire OAuth credentials onto gRPC stub
channel in camel-zeebe (#24695)
89f6a09c83da is described below
commit 89f6a09c83da0f3dceeecb7c4cfc4847819eb72b
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 7af75c65c535..64fbddeb8ef6 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;
@@ -52,6 +62,7 @@ public class ZeebeService {
private ZeebeClient zeebeClient;
private ManagedChannel managedChannel;
+ private Channel grpcChannel;
private ObjectMapper objectMapper;
@@ -78,15 +89,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)
@@ -102,6 +116,9 @@ public class ZeebeService {
managedChannel = ManagedChannelBuilder.forAddress(gatewayHost,
gatewayPort)
.usePlaintext()
.build();
+ grpcChannel = credentialsProvider != null
+ ? ClientInterceptors.intercept(managedChannel, new
OAuthInterceptor(credentialsProvider))
+ : managedChannel;
}
}
@@ -149,7 +166,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())
@@ -171,7 +188,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);
@@ -212,7 +229,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()) {
@@ -240,7 +257,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());
@@ -263,7 +280,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());
@@ -285,7 +302,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());
@@ -308,7 +325,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()))
@@ -361,4 +378,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);
+ }
+ };
+ }
+ }
}
