dapr / dapr/java-sdk

Actor state save best practices

Aperta
#854 8 commenti 0 reazioni 0 assegnatari Vedi su GitHub
kind/bug P1
Lingua principale
Java
Stelle
300
Fork
230
Merge medio
5g 1h
PR unite (30g)
5

Descrizione

## Expected Behavior

Saving actor state does not fail.

## Actual Behavior

Saving an actor state fails with the following:

```java
java.util.ConcurrentModificationException: null
at java.base/java.util.HashMap$HashIterator.nextNode(HashMap.java:1597)
at java.base/java.util.HashMap$EntryIterator.next(HashMap.java:1630)
at java.base/java.util.HashMap$EntryIterator.next(HashMap.java:1628)
at io.dapr.actors.runtime.ActorStateManager.flush(ActorStateManager.java:286)
at io.dapr.actors.runtime.ActorStateManager.lambda$save$15(ActorStateManager.java:272)
at reactor.core.publisher.MonoRunnable.call(MonoRunnable.java:73)
at reactor.core.publisher.MonoRunnable.call(MonoRunnable.java:32)
at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.subscribeNext(MonoIgnoreThen.java:228)
at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.onComplete(MonoIgnoreThen.java:203)
at reactor.core.publisher.FluxHide$SuppressFuseableSubscriber.onComplete(FluxHide.java:147)
at reactor.core.publisher.MonoIgnoreElements$IgnoreElementsSubscriber.onComplete(MonoIgnoreElements.java:89)
at reactor.core.publisher.MonoIgnoreElements$IgnoreElementsSubscriber.onComplete(MonoIgnoreElements.java:89)
at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1840)
at reactor.core.publisher.MonoCallable.subscribe(MonoCallable.java:62)
at reactor.core.publisher.Mono.subscribe(Mono.java:4490)
at reactor.core.publisher.MonoIgnorePublisher.subscribe(MonoIgnorePublisher.java:57)
at reactor.core.publisher.Mono.subscribe(Mono.java:4490)
at reactor.core.publisher.FluxFlatMap.trySubscribeScalarMap(FluxFlatMap.java:203)
at reactor.core.publisher.MonoFlatMap.subscribeOrReturn(MonoFlatMap.java:53)
at reactor.core.publisher.Mono.subscribe(Mono.java:4475)
at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.subscribeNext(MonoIgnoreThen.java:263)
at reactor.core.publisher.MonoIgnoreThen.subscribe(MonoIgnoreThen.java:51)
at reactor.core.publisher.Mono.subscribe(Mono.java:4490)
at reactor.core.publisher.Mono.block(Mono.java:1741)
```

## Steps to Reproduce the Problem

Implement an actor with two methods like the following:

```java
@Override
public void enqueue(String taskId) {

var mockPayload = "{}";

this.getActorStateManager().set("payload_%s".formatted(taskId), mockPayload).block();

System.out.println("Saved payload %s".formatted(taskId));

this.saveState().block();
}

@Override
public void ack(String taskId) {

var mockPayload = this.getActorStateManager().get("payload_%s".formatted(taskId), String.class).block();

System.out.println("Retrieved payload %s".formatted(taskId));
System.out.println("Payload was %s".formatted(mockPayload));

this.getActorStateManager().remove("payload_%s".formatted(taskId)).block();

this.saveState().block();
}
```

And play around, try to call:

1. `enqueue(1)`
2. `enqueue(2)`
3. `enqueue(3)`
4. `ack(1)`

Should fail.

Controller for the sake of completeness:

```java
@RestController
@RequiredArgsConstructor
public class SerializerActorController {

private final ActorProxyBuilder actorProxyBuilder;

@GetMapping(value = "/enqueue/{taskId}")
public ResponseEntity enqueue(@PathVariable("taskId") String taskId) {

try {
var actorId = new ActorId("actorId");
var actor = actorProxyBuilder.build(actorId);

actor.enqueue(taskId);

return ResponseEntity.ok().build();
} catch (Exception ex) {
ex.printStackTrace();
return ResponseEntity.internalServerError().build();
}
}

@GetMapping(value = "/ack/{taskId}")
public ResponseEntity ack(@PathVariable("taskId") String taskId) {

try {
var actorId = new ActorId("actorId");
var actor = actorProxyBuilder.build(actorId);

actor.ack(taskId);

return ResponseEntity.ok().build();
} catch (Exception ex) {
ex.printStackTrace();
return ResponseEntity.internalServerError().build();
}
}
}
```

## Release Note

RELEASE NOTE:

**FIX** Solved concurrency on actor state saving stage.

Guida per i contributori

Apri la guida per i contributori

Direzione di ricerca

Inizia riproducendo la sequenza enqueue(1), enqueue(2), enqueue(3) e ack(1), quindi ispeziona ActorStateManager.flush e save nelle posizioni dello stack trace. Conferma che il percorso di salvataggio dello stato dell’actor non sollevi più ConcurrentModificationException e che le operazioni sullo stato segnalate vengano completate correttamente.

Scritto dal modello di indicizzazione a partire dal testo della issue.

Valutazione

Stack tecnologico
java
Ambito
backend, distributed-systems
Tipo di issue
Bug
Difficoltà
4/5
Tempo stimato
3-5 giorni
Stato di attività
Ferma
Chiarezza
Abbastanza chiara
Idoneità per principianti
45/100

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.