Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 16 additions & 0 deletions examples/src/main/java/io/dapr/examples/actors/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -80,19 +81,21 @@ class DaprClientImpl implements DaprClient {
*/
@Override
public Mono<byte[]> 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.<DaprActorsProtos.InvokeActorResponse>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.<DaprActorsProtos.InvokeActorResponse>createMono(
it -> this.grpcInterceptors.intercept(client, context).invokeActor(request, it));
}).map(r -> r.getData().toByteArray());
}

private <T> Mono<T> createMono(Consumer<StreamObserver<T>> consumer) {
Expand Down
Original file line number Diff line number Diff line change
@@ -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<String> 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 <T> Publisher result type.
* @return The publisher with the invocation-scoped context value.
*/
public static <T> Mono<T> withReentrancyId(Mono<T> 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 <T> Invocation result type.
* @return The invocation result.
*/
public static <T> T withReentrancyId(String reentrancyId, Supplier<T> 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<String, String> 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);
}

}
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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()
Expand All @@ -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);
}

Expand Down Expand Up @@ -148,7 +156,9 @@ private static Mono<SubscribeActorEventsRequestAlpha1> 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()
Expand All @@ -167,9 +177,10 @@ private static Mono<SubscribeActorEventsRequestAlpha1> 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()))
Expand All @@ -183,10 +194,11 @@ private static Mono<SubscribeActorEventsRequestAlpha1> 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()))
Expand Down Expand Up @@ -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();
}
}
53 changes: 29 additions & 24 deletions sdk-actors/src/main/java/io/dapr/actors/runtime/ActorManager.java
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -260,20 +261,21 @@ private Mono<byte[]> invokeMethod(ActorId actorId, ActorMethodContext context, S
* @return Asynchronous object response.
*/
private Mono<Object> 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);
}
})));
}

/**
Expand All @@ -285,16 +287,19 @@ private Mono<Object> invokeMethod(AbstractActor actor, Method method, Object inp
* @return Asynchronous object response.
*/
private Mono<Object> invokeMonoMethod(AbstractActor actor, Method method, Object input) {
try {
if (method.getParameterCount() == 0) {
return (Mono<Object>) method.invoke(actor);
} else {
// Actor methods must have a one or no parameter, which is guaranteed at this point.
return (Mono<Object>) 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<Object>) method.invoke(actor);
} else {
// Actor methods must have a one or no parameter, which is guaranteed at this point.
return (Mono<Object>) method.invoke(actor, input);
}
} catch (Exception e) {
return Mono.error(e);
}
}));
}

/**
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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");
Expand Down Expand Up @@ -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();
}
Expand All @@ -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}
*/
Expand Down
Loading