diff --git a/examples/src/main/java/io/dapr/examples/actors/README.md b/examples/src/main/java/io/dapr/examples/actors/README.md index 4ed9e78c51..f56258b108 100644 --- a/examples/src/main/java/io/dapr/examples/actors/README.md +++ b/examples/src/main/java/io/dapr/examples/actors/README.md @@ -64,6 +64,22 @@ public class DemoActorService { ``` This application uses `ActorRuntime.getInstance().registerActor()` in order to register `DemoActorImpl` as an actor in the Dapr Actor runtime. Internally, it is using `DefaultObjectSerializer` for two properties: `objectSerializer` is for Dapr's sent and received objects, and `stateSerializer` is for objects to be persisted. + +### Enable actor reentrancy + +Reentrancy is disabled unless configured. It can be enabled for every actor type or overridden for a specific type. The same configuration is advertised by the HTTP actor host and the gRPC actor event stream: + +```java +ActorRuntime runtime = ActorRuntime.getInstance(); +runtime.getConfig().setReentrancy( + new ActorReentrancyConfig().setEnabled(true).setMaxStackDepth(5)); + +runtime.getConfig().addActorTypeConfig(new ActorTypeConfig() + .setActorTypeName("DemoActor") + .setReentrancy(new ActorReentrancyConfig().setEnabled(false))); +``` + +For reactive actor methods, compose nested calls through the returned `Mono` so Reactor context can carry Dapr's reentrancy id across scheduler changes and concurrent calls. Synchronous actor methods can make nested calls through a blocking actor proxy; the SDK scopes the reentrancy id to the method invocation. `DaprApplication.start()` method will run the Spring Boot [DaprApplication](https://github.com/dapr/java-sdk/blob/master/sdk-tests/src/test/java/io/dapr/it/actors/services/springboot/DaprApplication.java), which registers the Dapr Spring Boot controller [DaprController](https://github.com/dapr/java-sdk/blob/master/sdk-springboot/src/main/java/io/dapr/springboot/DaprController.java). This controller contains all Actor methods implemented as endpoints. The Dapr's sidecar will call into the controller. diff --git a/sdk-actors/src/main/java/io/dapr/actors/client/DaprClientImpl.java b/sdk-actors/src/main/java/io/dapr/actors/client/DaprClientImpl.java index 4e0fdfac4d..96f7fb6606 100644 --- a/sdk-actors/src/main/java/io/dapr/actors/client/DaprClientImpl.java +++ b/sdk-actors/src/main/java/io/dapr/actors/client/DaprClientImpl.java @@ -14,6 +14,7 @@ package io.dapr.actors.client; import com.google.protobuf.ByteString; +import io.dapr.actors.internal.ActorReentrancyContext; import io.dapr.client.resiliency.ResiliencyOptions; import io.dapr.exceptions.DaprException; import io.dapr.internal.grpc.DaprClientGrpcInterceptors; @@ -80,19 +81,21 @@ class DaprClientImpl implements DaprClient { */ @Override public Mono invoke(String actorType, String actorId, String methodName, byte[] jsonPayload) { - DaprActorsProtos.InvokeActorRequest req = - DaprActorsProtos.InvokeActorRequest.newBuilder() + return Mono.deferContextual(context -> { + DaprActorsProtos.InvokeActorRequest.Builder requestBuilder = DaprActorsProtos.InvokeActorRequest.newBuilder() .setActorType(actorType) .setActorId(actorId) .setMethod(methodName) .putAllMetadata(this.metadata) - .setData(jsonPayload == null ? ByteString.EMPTY : ByteString.copyFrom(jsonPayload)) - .build(); - return Mono.deferContextual( - context -> this.createMono( - it -> this.grpcInterceptors.intercept(client, context).invokeActor(req, it) - ) - ).map(r -> r.getData().toByteArray()); + .setData(jsonPayload == null ? ByteString.EMPTY : ByteString.copyFrom(jsonPayload)); + String reentrancyId = ActorReentrancyContext.getReentrancyId(context); + if (reentrancyId != null) { + requestBuilder.putMetadata(ActorReentrancyContext.REENTRANCY_ID_HEADER, reentrancyId); + } + DaprActorsProtos.InvokeActorRequest request = requestBuilder.build(); + return this.createMono( + it -> this.grpcInterceptors.intercept(client, context).invokeActor(request, it)); + }).map(r -> r.getData().toByteArray()); } private Mono createMono(Consumer> consumer) { diff --git a/sdk-actors/src/main/java/io/dapr/actors/internal/ActorReentrancyContext.java b/sdk-actors/src/main/java/io/dapr/actors/internal/ActorReentrancyContext.java new file mode 100644 index 0000000000..f407e18b9c --- /dev/null +++ b/sdk-actors/src/main/java/io/dapr/actors/internal/ActorReentrancyContext.java @@ -0,0 +1,106 @@ +/* + * Copyright 2026 The Dapr 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.dapr.actors.internal; + +import reactor.core.publisher.Mono; +import reactor.util.context.ContextView; + +import java.util.Map; +import java.util.function.Supplier; + +/** + * Internal helpers for carrying an incoming actor reentrancy id through a Reactor invocation. + */ +public final class ActorReentrancyContext { + + /** Metadata key used by Dapr to identify an actor reentrant call chain. */ + public static final String REENTRANCY_ID_HEADER = "Dapr-Reentrancy-Id"; + + private static final String CONTEXT_KEY = ActorReentrancyContext.class.getName() + ".id"; + + private static final ThreadLocal THREAD_LOCAL_REENTRANCY_ID = new ThreadLocal<>(); + + private ActorReentrancyContext() { + } + + /** + * Adds the reentrancy id to the subscriber context for the lifetime of this publisher. + * + * @param publisher Publisher for the actor callback. + * @param reentrancyId Id received from Dapr, or {@code null} when this is not a reentrant call. + * @param Publisher result type. + * @return The publisher with the invocation-scoped context value. + */ + public static Mono withReentrancyId(Mono publisher, String reentrancyId) { + if (reentrancyId == null || reentrancyId.isEmpty()) { + return publisher; + } + return publisher.contextWrite(context -> context.put(CONTEXT_KEY, reentrancyId)); + } + + /** + * Sets the reentrancy id for the duration of a synchronous actor method invocation. + * + * @param reentrancyId Id received from Dapr, or {@code null} when this is not a reentrant call. + * @param action Synchronous actor method invocation. + * @param Invocation result type. + * @return The invocation result. + */ + public static T withReentrancyId(String reentrancyId, Supplier action) { + String previous = THREAD_LOCAL_REENTRANCY_ID.get(); + if (reentrancyId == null || reentrancyId.isEmpty()) { + THREAD_LOCAL_REENTRANCY_ID.remove(); + } else { + THREAD_LOCAL_REENTRANCY_ID.set(reentrancyId); + } + + try { + return action.get(); + } finally { + if (previous == null) { + THREAD_LOCAL_REENTRANCY_ID.remove(); + } else { + THREAD_LOCAL_REENTRANCY_ID.set(previous); + } + } + } + + /** + * Reads the reentrancy id from a Reactor subscriber context. + * + * @param context Subscriber context. + * @return The id, or {@code null} when no actor invocation is active. + */ + public static String getReentrancyId(ContextView context) { + return context.hasKey(CONTEXT_KEY) ? context.get(CONTEXT_KEY) : THREAD_LOCAL_REENTRANCY_ID.get(); + } + + /** + * Reads the reentrancy id from callback metadata, ignoring header-name casing. + * + * @param metadata Callback metadata. + * @return The id, or {@code null} when Dapr did not send one. + */ + public static String getReentrancyId(Map metadata) { + if (metadata == null) { + return null; + } + return metadata.entrySet().stream() + .filter(entry -> REENTRANCY_ID_HEADER.equalsIgnoreCase(entry.getKey())) + .map(Map.Entry::getValue) + .findFirst() + .orElse(null); + } + +} diff --git a/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorEventDispatcher.java b/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorEventDispatcher.java index b0c6298c31..6179c9c482 100644 --- a/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorEventDispatcher.java +++ b/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorEventDispatcher.java @@ -19,7 +19,9 @@ import com.google.protobuf.ByteString; import com.google.protobuf.BytesValue; import com.google.protobuf.InvalidProtocolBufferException; +import io.dapr.actors.internal.ActorReentrancyContext; import io.dapr.v1.DaprActorsProtos.ActorEntityConfig; +import io.dapr.v1.DaprActorsProtos.ActorReentrancyConfig; import io.dapr.v1.DaprActorsProtos.SubscribeActorEventsRequestAlpha1; import io.dapr.v1.DaprActorsProtos.SubscribeActorEventsRequestDeactivateResponseAlpha1; import io.dapr.v1.DaprActorsProtos.SubscribeActorEventsRequestFailedAlpha1; @@ -70,6 +72,9 @@ static SubscribeActorEventsRequestAlpha1 buildInitialRequest(ActorRuntimeConfig if (config.getDrainBalancedActors() != null) { builder.setDrainRebalancedActors(config.getDrainBalancedActors()); } + if (config.getReentrancy() != null) { + builder.setReentrancy(toProto(config.getReentrancy())); + } for (ActorTypeConfig typeConfig : config.getActorTypeConfigs()) { ActorEntityConfig.Builder entityConfig = ActorEntityConfig.newBuilder() @@ -83,6 +88,9 @@ static SubscribeActorEventsRequestAlpha1 buildInitialRequest(ActorRuntimeConfig if (typeConfig.getDrainBalancedActors() != null) { entityConfig.setDrainRebalancedActors(typeConfig.getDrainBalancedActors()); } + if (typeConfig.getReentrancy() != null) { + entityConfig.setReentrancy(toProto(typeConfig.getReentrancy())); + } builder.addEntitiesConfig(entityConfig); } @@ -148,7 +156,9 @@ private static Mono invoke( byte[] payload = request.getData().isEmpty() ? null : request.getData().toByteArray(); String contentType = runtime.getObjectSerializer(request.getActorType()).getContentType(); - return runtime.invoke(request.getActorType(), request.getActorId(), request.getMethod(), payload) + return ActorReentrancyContext.withReentrancyId( + runtime.invoke(request.getActorType(), request.getActorId(), request.getMethod(), payload), + ActorReentrancyContext.getReentrancyId(request.getMetadataMap())) .map(ByteString::copyFrom) .defaultIfEmpty(ByteString.EMPTY) .map(data -> SubscribeActorEventsRequestAlpha1.newBuilder() @@ -167,9 +177,10 @@ private static Mono reminder( return Mono.just(notRegistered(id, request.getActorType())); } - return Mono.fromCallable(() -> buildCallbackBody( + return ActorReentrancyContext.withReentrancyId(Mono.fromCallable(() -> buildCallbackBody( null, request.getDueTime(), request.getPeriod(), request.hasData() ? request.getData() : null)) - .flatMap(body -> runtime.invokeReminder(request.getActorType(), request.getActorId(), request.getName(), body)) + .flatMap(body -> runtime.invokeReminder(request.getActorType(), request.getActorId(), request.getName(), body)), + ActorReentrancyContext.getReentrancyId(request.getMetadataMap())) .then(Mono.fromSupplier(() -> SubscribeActorEventsRequestAlpha1.newBuilder() .setReminderResponse(SubscribeActorEventsRequestReminderResponseAlpha1.newBuilder().setId(id)) .build())) @@ -183,10 +194,11 @@ private static Mono timer( return Mono.just(notRegistered(id, request.getActorType())); } - return Mono.fromCallable(() -> buildCallbackBody( + return ActorReentrancyContext.withReentrancyId(Mono.fromCallable(() -> buildCallbackBody( request.getCallback(), request.getDueTime(), request.getPeriod(), request.hasData() ? request.getData() : null)) - .flatMap(body -> runtime.invokeTimer(request.getActorType(), request.getActorId(), request.getName(), body)) + .flatMap(body -> runtime.invokeTimer(request.getActorType(), request.getActorId(), request.getName(), body)), + ActorReentrancyContext.getReentrancyId(request.getMetadataMap())) .then(Mono.fromSupplier(() -> SubscribeActorEventsRequestAlpha1.newBuilder() .setTimerResponse(SubscribeActorEventsRequestReminderResponseAlpha1.newBuilder().setId(id)) .build())) @@ -278,4 +290,12 @@ private static com.google.protobuf.Duration toProto(Duration duration) { .setNanos(duration.getNano()) .build(); } + + private static ActorReentrancyConfig toProto(io.dapr.actors.runtime.ActorReentrancyConfig config) { + ActorReentrancyConfig.Builder builder = ActorReentrancyConfig.newBuilder().setEnabled(config.isEnabled()); + if (config.getMaxStackDepth() != null) { + builder.setMaxStackDepth(config.getMaxStackDepth()); + } + return builder.build(); + } } diff --git a/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorManager.java b/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorManager.java index 303736cf57..71f9f06c22 100644 --- a/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorManager.java +++ b/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorManager.java @@ -14,6 +14,7 @@ package io.dapr.actors.runtime; import io.dapr.actors.ActorId; +import io.dapr.actors.internal.ActorReentrancyContext; import io.dapr.serializer.DaprObjectSerializer; import io.dapr.utils.TypeRef; import reactor.core.publisher.Mono; @@ -260,20 +261,21 @@ private Mono invokeMethod(ActorId actorId, ActorMethodContext context, S * @return Asynchronous object response. */ private Mono invokeMethod(AbstractActor actor, Method method, Object input) { - return Mono.fromSupplier(() -> { - try { - if (method.getParameterCount() == 0) { - return method.invoke(actor); - } else { - // Actor methods must have a one or no parameter, which is guaranteed at this point. - return method.invoke(actor, input); - } - } catch (RuntimeException e) { - throw e; - } catch (Exception e) { - throw new RuntimeException(e); - } - }); + return Mono.deferContextual(context -> Mono.fromSupplier(() -> + ActorReentrancyContext.withReentrancyId(ActorReentrancyContext.getReentrancyId(context), () -> { + try { + if (method.getParameterCount() == 0) { + return method.invoke(actor); + } else { + // Actor methods must have a one or no parameter, which is guaranteed at this point. + return method.invoke(actor, input); + } + } catch (RuntimeException e) { + throw e; + } catch (Exception e) { + throw new RuntimeException(e); + } + }))); } /** @@ -285,16 +287,19 @@ private Mono invokeMethod(AbstractActor actor, Method method, Object inp * @return Asynchronous object response. */ private Mono invokeMonoMethod(AbstractActor actor, Method method, Object input) { - try { - if (method.getParameterCount() == 0) { - return (Mono) method.invoke(actor); - } else { - // Actor methods must have a one or no parameter, which is guaranteed at this point. - return (Mono) method.invoke(actor, input); - } - } catch (Exception e) { - return Mono.error(e); - } + return Mono.deferContextual(context -> ActorReentrancyContext.withReentrancyId( + ActorReentrancyContext.getReentrancyId(context), () -> { + try { + if (method.getParameterCount() == 0) { + return (Mono) method.invoke(actor); + } else { + // Actor methods must have a one or no parameter, which is guaranteed at this point. + return (Mono) method.invoke(actor, input); + } + } catch (Exception e) { + return Mono.error(e); + } + })); } /** diff --git a/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorObjectSerializer.java b/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorObjectSerializer.java index cf78ea6810..99f61c7b5f 100644 --- a/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorObjectSerializer.java +++ b/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorObjectSerializer.java @@ -160,6 +160,9 @@ private byte[] serialize(ActorRuntimeConfig config) throws IOException { if (config.getRemindersStoragePartitions() != null) { generator.writeNumberField("remindersStoragePartitions", config.getRemindersStoragePartitions()); } + if (config.getReentrancy() != null) { + writeReentrancy(generator, config.getReentrancy()); + } if (!config.getActorTypeConfigs().isEmpty()) { generator.writeArrayFieldStart("entitiesConfig"); @@ -188,6 +191,9 @@ private byte[] serialize(ActorRuntimeConfig config) throws IOException { if (actorTypeConfig.getRemindersStoragePartitions() != null) { generator.writeNumberField("remindersStoragePartitions", actorTypeConfig.getRemindersStoragePartitions()); } + if (actorTypeConfig.getReentrancy() != null) { + writeReentrancy(generator, actorTypeConfig.getReentrancy()); + } generator.writeEndObject(); } @@ -200,6 +206,15 @@ private byte[] serialize(ActorRuntimeConfig config) throws IOException { } } + private void writeReentrancy(JsonGenerator generator, ActorReentrancyConfig config) throws IOException { + generator.writeObjectFieldStart("reentrancy"); + generator.writeBooleanField("enabled", config.isEnabled()); + if (config.getMaxStackDepth() != null) { + generator.writeNumberField("maxStackDepth", config.getMaxStackDepth()); + } + generator.writeEndObject(); + } + /** * {@inheritDoc} */ diff --git a/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorReentrancyConfig.java b/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorReentrancyConfig.java new file mode 100644 index 0000000000..cff9bee683 --- /dev/null +++ b/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorReentrancyConfig.java @@ -0,0 +1,70 @@ +/* + * Copyright 2026 The Dapr 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.dapr.actors.runtime; + +/** + * Configures reentrant actor calls. + */ +public class ActorReentrancyConfig { + + private boolean enabled; + + private Integer maxStackDepth; + + /** + * Creates an actor reentrancy configuration. Reentrancy is disabled by default. + */ + public ActorReentrancyConfig() { + } + + /** + * Gets whether actor reentrancy is enabled. + * + * @return {@code true} when reentrant calls are allowed. + */ + public boolean isEnabled() { + return enabled; + } + + /** + * Sets whether actor reentrancy is enabled. + * + * @param enabled Whether actor reentrancy is enabled. + * @return This instance. + */ + public ActorReentrancyConfig setEnabled(boolean enabled) { + this.enabled = enabled; + return this; + } + + /** + * Gets the maximum number of nested reentrant calls. + * + * @return Maximum reentrancy stack depth, or {@code null} to use Dapr's default. + */ + public Integer getMaxStackDepth() { + return maxStackDepth; + } + + /** + * Sets the maximum number of nested reentrant calls. + * + * @param maxStackDepth Maximum reentrancy stack depth, or {@code null} to use Dapr's default. + * @return This instance. + */ + public ActorReentrancyConfig setMaxStackDepth(Integer maxStackDepth) { + this.maxStackDepth = maxStackDepth; + return this; + } +} diff --git a/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorRuntimeConfig.java b/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorRuntimeConfig.java index 7e42d49ae6..944d54194a 100644 --- a/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorRuntimeConfig.java +++ b/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorRuntimeConfig.java @@ -36,6 +36,8 @@ public class ActorRuntimeConfig { private volatile Integer remindersStoragePartitions; + private volatile ActorReentrancyConfig reentrancy; + private volatile List actorTypeConfigs = new CopyOnWriteArrayList<>(); /** @@ -169,12 +171,32 @@ public ActorRuntimeConfig setRemindersStoragePartitions(Integer remindersStorage } /** - * Adds a actor type config to the list of actors type configs. - * - * @param actorTypeName Actor type that was registered. + * Gets the default reentrancy configuration for all actor types. + * + * @return Default reentrancy configuration, or {@code null} to use Dapr's default. + */ + public ActorReentrancyConfig getReentrancy() { + return reentrancy; + } + + /** + * Sets the default reentrancy configuration for all actor types. + * + * @param reentrancy Default reentrancy configuration, or {@code null} to use Dapr's default. + * @return This instance. + */ + public ActorRuntimeConfig setReentrancy(ActorReentrancyConfig reentrancy) { + this.reentrancy = reentrancy; + return this; + } + + /** + * Adds an actor type config to the list of actor type configs. + * + * @param config Actor type config to add. * @return This instance. */ - ActorRuntimeConfig addActorTypeConfig(ActorTypeConfig config) { + public ActorRuntimeConfig addActorTypeConfig(ActorTypeConfig config) { if (config == null) { throw new IllegalArgumentException("Add actor type config failed."); } diff --git a/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorTypeConfig.java b/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorTypeConfig.java index ce3d2d524e..da6bf1ae7f 100644 --- a/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorTypeConfig.java +++ b/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorTypeConfig.java @@ -32,10 +32,12 @@ public class ActorTypeConfig { private volatile Integer remindersStoragePartitions; + private volatile ActorReentrancyConfig reentrancy; + /** * Instantiates a new config for the Actor Runtime. */ - ActorTypeConfig() { + public ActorTypeConfig() { } /** @@ -44,7 +46,7 @@ public class ActorTypeConfig { * @param actorTypeName Actor type that was registered. * @return This instance. */ - ActorTypeConfig setActorTypeName(String actorTypeName) { + public ActorTypeConfig setActorTypeName(String actorTypeName) { if (actorTypeName == null) { throw new IllegalArgumentException("Registered actor must have a type name."); } @@ -161,4 +163,24 @@ public ActorTypeConfig setRemindersStoragePartitions(Integer remindersStoragePar this.remindersStoragePartitions = remindersStoragePartitions; return this; } + + /** + * Gets this actor type's reentrancy configuration. + * + * @return Reentrancy configuration, or {@code null} to inherit the runtime default. + */ + public ActorReentrancyConfig getReentrancy() { + return reentrancy; + } + + /** + * Sets this actor type's reentrancy configuration. + * + * @param reentrancy Reentrancy configuration, or {@code null} to inherit the runtime default. + * @return This instance. + */ + public ActorTypeConfig setReentrancy(ActorReentrancyConfig reentrancy) { + this.reentrancy = reentrancy; + return this; + } } diff --git a/sdk-actors/src/test/java/io/dapr/actors/client/DaprGrpcClientTest.java b/sdk-actors/src/test/java/io/dapr/actors/client/DaprGrpcClientTest.java index e2f75a2b41..b56babfeff 100644 --- a/sdk-actors/src/test/java/io/dapr/actors/client/DaprGrpcClientTest.java +++ b/sdk-actors/src/test/java/io/dapr/actors/client/DaprGrpcClientTest.java @@ -14,6 +14,7 @@ package io.dapr.actors.client; import com.google.protobuf.ByteString; +import io.dapr.actors.internal.ActorReentrancyContext; import io.dapr.v1.DaprActorsProtos; import io.dapr.v1.DaprGrpc; import io.grpc.ManagedChannel; @@ -29,6 +30,8 @@ import reactor.core.publisher.Mono; import java.io.IOException; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutionException; import static io.dapr.actors.TestUtils.assertThrowsDaprException; @@ -53,6 +56,8 @@ public class DaprGrpcClientTest { private static final byte[] RESPONSE_PAYLOAD = "\"OK\"".getBytes(); + private final Set observedReentrancyIds = ConcurrentHashMap.newKeySet(); + @Rule public final GrpcCleanupRule grpcCleanup = new GrpcCleanupRule(); @@ -67,6 +72,11 @@ public void invokeActor(DaprActorsProtos.InvokeActorRequest request, switch (request.getActorId()) { case ACTOR_ID_OK: assertArrayEquals(REQUEST_PAYLOAD, request.getData().toByteArray()); + String reentrancyId = request.getMetadataMap() + .get(ActorReentrancyContext.REENTRANCY_ID_HEADER); + if (reentrancyId != null) { + observedReentrancyIds.add(reentrancyId); + } responseObserver.onNext( DaprActorsProtos.InvokeActorResponse.newBuilder().setData(ByteString.copyFrom(RESPONSE_PAYLOAD)) .build()); @@ -115,6 +125,18 @@ public void invoke() { assertArrayEquals(RESPONSE_PAYLOAD, result.block()); } + @Test + public void invokeUsesReentrancyIdFromEachSubscriberContext() { + Mono.when( + ActorReentrancyContext.withReentrancyId( + client.invoke(ACTOR_TYPE, ACTOR_ID_OK, METHOD_NAME, REQUEST_PAYLOAD), "chain-a"), + ActorReentrancyContext.withReentrancyId( + client.invoke(ACTOR_TYPE, ACTOR_ID_OK, METHOD_NAME, REQUEST_PAYLOAD), "chain-b")) + .block(); + + org.junit.jupiter.api.Assertions.assertEquals(Set.of("chain-a", "chain-b"), observedReentrancyIds); + } + @Test public void invokeNullPayload() { Mono result = client.invoke(ACTOR_TYPE, ACTOR_ID_NULL_INPUT, METHOD_NAME, null); diff --git a/sdk-actors/src/test/java/io/dapr/actors/internal/ActorReentrancyContextTest.java b/sdk-actors/src/test/java/io/dapr/actors/internal/ActorReentrancyContextTest.java new file mode 100644 index 0000000000..e3dd3c09dd --- /dev/null +++ b/sdk-actors/src/test/java/io/dapr/actors/internal/ActorReentrancyContextTest.java @@ -0,0 +1,58 @@ +/* + * Copyright 2026 The Dapr 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.dapr.actors.internal; + +import org.junit.jupiter.api.Test; +import reactor.core.publisher.Mono; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; + +class ActorReentrancyContextTest { + + @Test + void synchronousScopesAreVisibleToBlockingCallsAndRestored() { + String result = ActorReentrancyContext.withReentrancyId("outer", () -> { + assertEquals("outer", Mono.deferContextual(context -> + Mono.justOrEmpty(ActorReentrancyContext.getReentrancyId(context))).block()); + + String nestedResult = ActorReentrancyContext.withReentrancyId("inner", () -> + Mono.deferContextual(context -> Mono.justOrEmpty( + ActorReentrancyContext.getReentrancyId(context))).block()); + assertEquals("inner", nestedResult); + + return Mono.deferContextual(context -> Mono.justOrEmpty( + ActorReentrancyContext.getReentrancyId(context))).block(); + }); + + assertEquals("outer", result); + assertNull(Mono.deferContextual(context -> Mono.justOrEmpty( + ActorReentrancyContext.getReentrancyId(context))).block()); + } + + @Test + void nestedNonReentrantCallDoesNotInheritTheOuterId() { + String result = ActorReentrancyContext.withReentrancyId("outer", () -> { + String nestedResult = ActorReentrancyContext.withReentrancyId(null, () -> + Mono.deferContextual(context -> Mono.justOrEmpty( + ActorReentrancyContext.getReentrancyId(context))).block()); + assertNull(nestedResult); + return Mono.deferContextual(context -> Mono.justOrEmpty( + ActorReentrancyContext.getReentrancyId(context))).block(); + }); + + assertEquals("outer", result); + } + +} diff --git a/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorEventDispatcherTest.java b/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorEventDispatcherTest.java index fa59799994..531aec439e 100644 --- a/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorEventDispatcherTest.java +++ b/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorEventDispatcherTest.java @@ -18,6 +18,7 @@ import com.google.protobuf.BytesValue; import io.dapr.actors.ActorId; import io.dapr.actors.ActorType; +import io.dapr.actors.internal.ActorReentrancyContext; import io.dapr.serializer.DefaultObjectSerializer; import io.dapr.utils.TypeRef; import io.dapr.v1.DaprActorsProtos.ActorEntityConfig; @@ -106,6 +107,24 @@ public void lookup() throws NoSuchMethodException { } } + public interface ReentrancyActor { + Mono currentReentrancyId(); + } + + @ActorType(name = "ReentrancyActor") + public static class ReentrancyActorImpl extends AbstractActor implements ReentrancyActor { + + public ReentrancyActorImpl(ActorRuntimeContext runtimeContext, ActorId id) { + super(runtimeContext, id); + } + + @Override + public Mono currentReentrancyId() { + return Mono.deferContextual(context -> + Mono.justOrEmpty(ActorReentrancyContext.getReentrancyId(context))); + } + } + private ActorRuntime runtime; static ActorRuntime newRuntime() throws Exception { @@ -128,6 +147,7 @@ public void setup() throws Exception { runtime.registerActor(ActorRuntimeTest.MyActorImpl.class); runtime.registerActor(MyRemindableActorImpl.class); runtime.registerActor(ThrowingActorImpl.class); + runtime.registerActor(ReentrancyActorImpl.class); } @Test @@ -136,9 +156,11 @@ public void initialRequestCarriesTypesAndConfig() { .setActorIdleTimeout(Duration.ofMinutes(2)) .setDrainOngoingCallTimeout(Duration.ofMillis(1500)) .setDrainBalancedActors(true) + .setReentrancy(new io.dapr.actors.runtime.ActorReentrancyConfig().setEnabled(true).setMaxStackDepth(5)) .addActorTypeConfig(new ActorTypeConfig() .setActorTypeName("MyGreatActor") - .setActorIdleTimeout(Duration.ofSeconds(30))) + .setActorIdleTimeout(Duration.ofSeconds(30)) + .setReentrancy(new io.dapr.actors.runtime.ActorReentrancyConfig().setEnabled(false).setMaxStackDepth(2))) .addActorTypeConfig(new ActorTypeConfig() .setActorTypeName("MyRemindableActor") .setDrainOngoingCallTimeout(Duration.ofSeconds(5)) @@ -147,15 +169,21 @@ public void initialRequestCarriesTypesAndConfig() { SubscribeActorEventsRequestInitialAlpha1 initial = ActorEventDispatcher.buildInitialRequest(runtime.getConfig()).getInitialRequest(); - assertEquals(Arrays.asList("MyGreatActor", "MyRemindableActor", "ThrowingActor"), initial.getEntitiesList()); + assertEquals(Arrays.asList("MyGreatActor", "MyRemindableActor", "ThrowingActor", "ReentrancyActor"), + initial.getEntitiesList()); assertEquals(120, initial.getActorIdleTimeout().getSeconds()); assertEquals(1, initial.getDrainOngoingCallTimeout().getSeconds()); assertEquals(500_000_000, initial.getDrainOngoingCallTimeout().getNanos()); assertTrue(initial.getDrainRebalancedActors()); - assertFalse(initial.hasReentrancy()); + assertTrue(initial.hasReentrancy()); + assertTrue(initial.getReentrancy().getEnabled()); + assertEquals(5, initial.getReentrancy().getMaxStackDepth()); ActorEntityConfig entityConfig = initial.getEntitiesConfig(0); assertEquals(Arrays.asList("MyGreatActor"), entityConfig.getEntitiesList()); assertEquals(30, entityConfig.getActorIdleTimeout().getSeconds()); + assertTrue(entityConfig.hasReentrancy()); + assertFalse(entityConfig.getReentrancy().getEnabled()); + assertEquals(2, entityConfig.getReentrancy().getMaxStackDepth()); assertFalse(entityConfig.hasDrainOngoingCallTimeout()); ActorEntityConfig secondConfig = initial.getEntitiesConfig(1); assertFalse(secondConfig.hasActorIdleTimeout()); @@ -190,6 +218,24 @@ public void invokePassesData() throws Exception { assertEquals("payload", LAST_TIMER_DATA.get()); } + @Test + public void invokeCarriesReentrancyIdInInvocationContext() throws Exception { + String reentrancyId = "chain-123"; + String actorId = UUID.randomUUID().toString(); + SubscribeActorEventsResponseAlpha1 request = SubscribeActorEventsResponseAlpha1.newBuilder() + .setInvokeRequest(SubscribeActorEventsResponseInvokeRequestAlpha1.newBuilder() + .setId("reentrant") + .setActorType("ReentrancyActor") + .setActorId(actorId) + .setMethod("currentReentrancyId") + .putMetadata(ActorReentrancyContext.REENTRANCY_ID_HEADER, reentrancyId)) + .build(); + + SubscribeActorEventsRequestAlpha1 response = dispatch(request); + + assertEquals(reentrancyId, SERIALIZER.deserialize(response.getInvokeResponse().getData().toByteArray(), String.class)); + } + @Test public void callbacksForUnknownActorTypeAreNotFound() { String actorId = UUID.randomUUID().toString(); @@ -217,6 +263,7 @@ public void initialRequestLeavesUnsetConfigUnset() { assertFalse(initial.hasActorIdleTimeout()); assertFalse(initial.hasDrainOngoingCallTimeout()); assertFalse(initial.hasDrainRebalancedActors()); + assertFalse(initial.hasReentrancy()); } @Test diff --git a/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorObjectSerializerTest.java b/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorObjectSerializerTest.java index a5178bf035..d7f3603c48 100644 --- a/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorObjectSerializerTest.java +++ b/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorObjectSerializerTest.java @@ -149,7 +149,8 @@ public void serializeActorRuntimeConfigWithAllAttributes() throws IOException { .setActorScanInterval(Duration.ofSeconds(30)) .setDrainOngoingCallTimeout(Duration.ofMinutes(1)) .setDrainBalancedActors(true) - .setRemindersStoragePartitions(7); + .setRemindersStoragePartitions(7) + .setReentrancy(new ActorReentrancyConfig().setEnabled(true).setMaxStackDepth(5)); byte[] serialized = SERIALIZER.serialize(config); @@ -158,7 +159,8 @@ public void serializeActorRuntimeConfigWithAllAttributes() throws IOException { + "\"actorScanInterval\":\"0h0m30s0ms\"," + "\"drainOngoingCallTimeout\":\"0h1m0s0ms\"," + "\"drainBalancedActors\":true," - + "\"remindersStoragePartitions\":7}"; + + "\"remindersStoragePartitions\":7," + + "\"reentrancy\":{\"enabled\":true,\"maxStackDepth\":5}}"; Assertions.assertEquals(OBJECT_MAPPER.readTree(expected), OBJECT_MAPPER.readTree(serialized)); } @@ -168,7 +170,8 @@ public void serializeActorRuntimeConfigWithActorTypeConfig() throws IOException .setActorTypeName("MyActor") .setActorIdleTimeout(Duration.ofHours(2)) .setDrainBalancedActors(false) - .setRemindersStoragePartitions(3); + .setRemindersStoragePartitions(3) + .setReentrancy(new ActorReentrancyConfig().setEnabled(false).setMaxStackDepth(2)); ActorRuntimeConfig config = new ActorRuntimeConfig() .addRegisteredActorType("MyActor") .addActorTypeConfig(actorTypeConfig); @@ -180,7 +183,8 @@ public void serializeActorRuntimeConfigWithActorTypeConfig() throws IOException + "\"entities\":[\"MyActor\"]," + "\"actorIdleTimeout\":\"2h0m0s0ms\"," + "\"drainBalancedActors\":false," - + "\"remindersStoragePartitions\":3}]}"; + + "\"remindersStoragePartitions\":3," + + "\"reentrancy\":{\"enabled\":false,\"maxStackDepth\":2}}]}"; Assertions.assertEquals(OBJECT_MAPPER.readTree(expected), OBJECT_MAPPER.readTree(serialized)); } diff --git a/sdk-springboot/src/main/java/io/dapr/springboot/DaprController.java b/sdk-springboot/src/main/java/io/dapr/springboot/DaprController.java index 35794e826d..c283585549 100644 --- a/sdk-springboot/src/main/java/io/dapr/springboot/DaprController.java +++ b/sdk-springboot/src/main/java/io/dapr/springboot/DaprController.java @@ -13,6 +13,7 @@ package io.dapr.springboot; +import io.dapr.actors.internal.ActorReentrancyContext; import io.dapr.actors.runtime.ActorRuntime; import io.dapr.serializer.DefaultObjectSerializer; import org.springframework.http.MediaType; @@ -21,6 +22,7 @@ import org.springframework.web.bind.annotation.PathVariable; import org.springframework.web.bind.annotation.PutMapping; import org.springframework.web.bind.annotation.RequestBody; +import org.springframework.web.bind.annotation.RequestHeader; import org.springframework.web.bind.annotation.RestController; import reactor.core.publisher.Mono; @@ -82,14 +84,18 @@ public Mono deactivateActor(@PathVariable("type") String type, * @param id Actor Id. * @param method Actor method. * @param body Raw request body. + * @param reentrancyId Dapr reentrancy id, if present. * @return Raw response body. */ @PutMapping(path = "/actors/{type}/{id}/method/{method}") public Mono invokeActorMethod(@PathVariable("type") String type, @PathVariable("id") String id, @PathVariable("method") String method, - @RequestBody(required = false) byte[] body) { - return ActorRuntime.getInstance().invoke(type, id, method, body); + @RequestBody(required = false) byte[] body, + @RequestHeader(value = ActorReentrancyContext.REENTRANCY_ID_HEADER, + required = false) String reentrancyId) { + return ActorReentrancyContext.withReentrancyId( + ActorRuntime.getInstance().invoke(type, id, method, body), reentrancyId); } /** @@ -98,14 +104,18 @@ public Mono invokeActorMethod(@PathVariable("type") String type, * @param id Actor Id. * @param timer Actor timer's name. * @param body Raw request's body. + * @param reentrancyId Dapr reentrancy id, if present. * @return Void. */ @PutMapping(path = "/actors/{type}/{id}/method/timer/{timer}") public Mono invokeActorTimer(@PathVariable("type") String type, @PathVariable("id") String id, @PathVariable("timer") String timer, - @RequestBody byte[] body) { - return ActorRuntime.getInstance().invokeTimer(type, id, timer, body); + @RequestBody byte[] body, + @RequestHeader(value = ActorReentrancyContext.REENTRANCY_ID_HEADER, + required = false) String reentrancyId) { + return ActorReentrancyContext.withReentrancyId( + ActorRuntime.getInstance().invokeTimer(type, id, timer, body), reentrancyId); } /** @@ -114,13 +124,17 @@ public Mono invokeActorTimer(@PathVariable("type") String type, * @param id Actor Id. * @param reminder Actor reminder's name. * @param body Raw request's body. + * @param reentrancyId Dapr reentrancy id, if present. * @return Void. */ @PutMapping(path = "/actors/{type}/{id}/method/remind/{reminder}") public Mono invokeActorReminder(@PathVariable("type") String type, @PathVariable("id") String id, @PathVariable("reminder") String reminder, - @RequestBody(required = false) byte[] body) { - return ActorRuntime.getInstance().invokeReminder(type, id, reminder, body); + @RequestBody(required = false) byte[] body, + @RequestHeader(value = ActorReentrancyContext.REENTRANCY_ID_HEADER, + required = false) String reentrancyId) { + return ActorReentrancyContext.withReentrancyId( + ActorRuntime.getInstance().invokeReminder(type, id, reminder, body), reentrancyId); } } diff --git a/sdk-tests/src/test/java/io/dapr/it/actors/ActorReentrancyIT.java b/sdk-tests/src/test/java/io/dapr/it/actors/ActorReentrancyIT.java new file mode 100644 index 0000000000..ba8ec3bfc1 --- /dev/null +++ b/sdk-tests/src/test/java/io/dapr/it/actors/ActorReentrancyIT.java @@ -0,0 +1,109 @@ +/* + * Copyright 2026 The Dapr 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.dapr.it.actors; + +import io.dapr.actors.ActorId; +import io.dapr.actors.client.ActorClient; +import io.dapr.actors.client.ActorProxyBuilder; +import io.dapr.it.AppRun; +import io.dapr.it.actors.app.MyActorGrpcService; +import io.dapr.it.actors.app.MyActorService; +import io.dapr.it.actors.app.ReentrantActor; +import io.dapr.it.actors.app.ReentrantActorImpl; +import io.dapr.it.containers.BaseContainerIT; +import io.dapr.testcontainers.DaprContainer; +import io.dapr.testcontainers.DaprProtocol; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; + +import java.util.UUID; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** Verifies that an incoming reentrancy id is preserved for nested actor calls over both host protocols. */ +public class ActorReentrancyIT extends BaseContainerIT { + + private static DaprContainer httpDapr; + + private static DaprContainer grpcDapr; + + private static ActorClient httpActorClient; + + private static ActorClient grpcActorClient; + + @BeforeAll + public static void start() throws Exception { + var httpPair = startAppAndAttach( + "actor-reentrancy-http-it", + MyActorService.class, + AppRun.AppProtocol.HTTP, + appPort -> daprBuilder("actor-reentrancy-http-it") + .withAppPort(appPort) + .withAppChannelAddress("host.testcontainers.internal") + .withComponent(redisStateStore(STATE_STORE_NAME))); + httpDapr = httpPair.dapr(); + httpActorClient = newActorClient(httpDapr); + waitForActorsReady(httpDapr); + + var grpcPair = startActorHostAndAttach( + "actor-reentrancy-grpc-it", + MyActorGrpcService.class, + () -> daprBuilder("actor-reentrancy-grpc-it") + .withAppProtocol(DaprProtocol.GRPC) + .withComponent(redisStateStore(STATE_STORE_NAME))); + grpcDapr = grpcPair.dapr(); + grpcActorClient = newActorClient(grpcDapr); + waitForActorsReady(grpcDapr); + } + + @Test + public void reentersActorOverHttpHost() { + assertReentrantCallCompletes(httpActorClient); + } + + @Test + public void reentersActorSynchronouslyOverHttpHost() { + assertSynchronousReentrantCallCompletes(httpActorClient); + } + + @Test + public void reentersActorOverGrpcHost() { + assertReentrantCallCompletes(grpcActorClient); + } + + @Test + public void reentersActorSynchronouslyOverGrpcHost() { + assertSynchronousReentrantCallCompletes(grpcActorClient); + } + + private static void assertReentrantCallCompletes(ActorClient actorClient) { + ActorProxyBuilder builder = + new ActorProxyBuilder<>(ReentrantActorImpl.TYPE, ReentrantActor.class, actorClient); + String actorAId = UUID.randomUUID().toString(); + String actorBId = UUID.randomUUID().toString(); + ReentrantActor actorA = builder.build(new ActorId(actorAId)); + + assertEquals("reentrancy-ok", actorA.call(actorBId).block()); + } + + private static void assertSynchronousReentrantCallCompletes(ActorClient actorClient) { + ActorProxyBuilder builder = + new ActorProxyBuilder<>(ReentrantActorImpl.TYPE, ReentrantActor.class, actorClient); + String actorAId = UUID.randomUUID().toString(); + String actorBId = UUID.randomUUID().toString(); + ReentrantActor actorA = builder.build(new ActorId(actorAId)); + + assertEquals("sync-reentrancy-ok", actorA.callBlocking(actorBId)); + } +} diff --git a/sdk-tests/src/test/java/io/dapr/it/actors/app/MyActorGrpcService.java b/sdk-tests/src/test/java/io/dapr/it/actors/app/MyActorGrpcService.java index cd17f07750..a097f96a49 100644 --- a/sdk-tests/src/test/java/io/dapr/it/actors/app/MyActorGrpcService.java +++ b/sdk-tests/src/test/java/io/dapr/it/actors/app/MyActorGrpcService.java @@ -15,6 +15,8 @@ import io.dapr.actors.runtime.ActorEventSubscription; import io.dapr.actors.runtime.ActorRuntime; +import io.dapr.actors.runtime.ActorReentrancyConfig; +import io.dapr.actors.runtime.ActorTypeConfig; import java.time.Duration; @@ -35,6 +37,10 @@ public static void main(String[] args) throws Exception { ActorRuntime.getInstance().registerActor(MyActorStringImpl.class); ActorRuntime.getInstance().registerActor(MyActorBinaryImpl.class); ActorRuntime.getInstance().registerActor(MyActorObjectImpl.class); + ActorRuntime.getInstance().getConfig().addActorTypeConfig(new ActorTypeConfig() + .setActorTypeName(ReentrantActorImpl.TYPE) + .setReentrancy(new ActorReentrancyConfig().setEnabled(true).setMaxStackDepth(5))); + ActorRuntime.getInstance().registerActor(ReentrantActorImpl.class); try (ActorEventSubscription subscription = ActorRuntime.getInstance().subscribeActorEvents()) { subscription.awaitTermination(); diff --git a/sdk-tests/src/test/java/io/dapr/it/actors/app/MyActorService.java b/sdk-tests/src/test/java/io/dapr/it/actors/app/MyActorService.java index 94f9959822..b1ed79f4ee 100644 --- a/sdk-tests/src/test/java/io/dapr/it/actors/app/MyActorService.java +++ b/sdk-tests/src/test/java/io/dapr/it/actors/app/MyActorService.java @@ -14,6 +14,8 @@ package io.dapr.it.actors.app; import io.dapr.actors.runtime.ActorRuntime; +import io.dapr.actors.runtime.ActorReentrancyConfig; +import io.dapr.actors.runtime.ActorTypeConfig; public class MyActorService { public static final String SUCCESS_MESSAGE = "dapr initialized. Status: Running"; @@ -30,6 +32,10 @@ public static void main(String[] args) throws Exception { ActorRuntime.getInstance().registerActor(MyActorStringImpl.class); ActorRuntime.getInstance().registerActor(MyActorBinaryImpl.class); ActorRuntime.getInstance().registerActor(MyActorObjectImpl.class); + ActorRuntime.getInstance().getConfig().addActorTypeConfig(new ActorTypeConfig() + .setActorTypeName(ReentrantActorImpl.TYPE) + .setReentrancy(new ActorReentrancyConfig().setEnabled(true).setMaxStackDepth(5))); + ActorRuntime.getInstance().registerActor(ReentrantActorImpl.class); TestApplication.start(port); } diff --git a/sdk-tests/src/test/java/io/dapr/it/actors/app/ReentrantActor.java b/sdk-tests/src/test/java/io/dapr/it/actors/app/ReentrantActor.java new file mode 100644 index 0000000000..ba389a9b76 --- /dev/null +++ b/sdk-tests/src/test/java/io/dapr/it/actors/app/ReentrantActor.java @@ -0,0 +1,32 @@ +/* + * Copyright 2026 The Dapr 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.dapr.it.actors.app; + +import reactor.core.publisher.Mono; + +/** Actor API used by the actor reentrancy integration test. */ +public interface ReentrantActor { + + Mono call(String targetActorId); + + Mono returnTo(String actorId); + + Mono finish(); + + String callBlocking(String targetActorId); + + String returnToBlocking(String actorId); + + String finishBlocking(); +} diff --git a/sdk-tests/src/test/java/io/dapr/it/actors/app/ReentrantActorImpl.java b/sdk-tests/src/test/java/io/dapr/it/actors/app/ReentrantActorImpl.java new file mode 100644 index 0000000000..b451d9abbd --- /dev/null +++ b/sdk-tests/src/test/java/io/dapr/it/actors/app/ReentrantActorImpl.java @@ -0,0 +1,72 @@ +/* + * Copyright 2026 The Dapr 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.dapr.it.actors.app; + +import io.dapr.actors.ActorId; +import io.dapr.actors.ActorType; +import io.dapr.actors.client.ActorClient; +import io.dapr.actors.client.ActorProxyBuilder; +import io.dapr.actors.runtime.AbstractActor; +import io.dapr.actors.runtime.ActorRuntimeContext; +import reactor.core.publisher.Mono; + +/** Actor whose call method performs an A -> B -> A actor call chain. */ +@ActorType(name = ReentrantActorImpl.TYPE) +public class ReentrantActorImpl extends AbstractActor implements ReentrantActor { + + public static final String TYPE = "ReentrantActorTest"; + + private static final ActorClient ACTOR_CLIENT = new ActorClient(); + + private static final ActorProxyBuilder PROXY_BUILDER = + new ActorProxyBuilder<>(TYPE, ReentrantActor.class, ACTOR_CLIENT); + + public ReentrantActorImpl(ActorRuntimeContext runtimeContext, ActorId id) { + super(runtimeContext, id); + } + + @Override + public Mono call(String targetActorId) { + return proxy(targetActorId).returnTo(getId().toString()); + } + + @Override + public Mono returnTo(String actorId) { + return proxy(actorId).finish(); + } + + @Override + public Mono finish() { + return Mono.just("reentrancy-ok"); + } + + @Override + public String callBlocking(String targetActorId) { + return proxy(targetActorId).returnToBlocking(getId().toString()); + } + + @Override + public String returnToBlocking(String actorId) { + return proxy(actorId).finishBlocking(); + } + + @Override + public String finishBlocking() { + return "sync-reentrancy-ok"; + } + + private static ReentrantActor proxy(String actorId) { + return PROXY_BUILDER.build(new ActorId(actorId)); + } +}