Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix: logging workflow step and transition failures #146

Merged
merged 1 commit into from
Jan 14, 2025
Merged

fix: logging workflow step and transition failures #146

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
1 change: 1 addition & 0 deletions akka-javasdk/src/main/scala/akka/javasdk/impl/SdkRunner.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -381,6 +381,7 @@ private final class Sdk(
logger.debug(s"Registering Workflow [${clz.getName}]")
new WorkflowImpl[S, W](
factoryContext.workflowId,
clz,
serializer,
ComponentDescriptor.descriptorFor(clz, serializer),
timerClient = runtimeComponentClients.timerClient,
Expand Down
57 changes: 20 additions & 37 deletions akka-javasdk/src/main/scala/akka/javasdk/impl/workflow/ReflectiveWorkflowRouter.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,9 +16,7 @@ import akka.annotation.InternalApi
import akka.javasdk.impl.MethodInvoker
import akka.javasdk.impl.CommandSerialization
import akka.javasdk.impl.HandlerNotFoundException
import akka.javasdk.impl.WorkflowExceptions.WorkflowException
import akka.javasdk.impl.serialization.JsonSerializer
import akka.javasdk.impl.workflow.ReflectiveWorkflowRouter.CommandHandlerNotFound
import akka.javasdk.impl.workflow.ReflectiveWorkflowRouter.CommandResult
import akka.javasdk.impl.workflow.ReflectiveWorkflowRouter.TransitionalResult
import akka.javasdk.impl.workflow.ReflectiveWorkflowRouter.WorkflowStepNotFound
Expand All @@ -40,9 +38,6 @@ object ReflectiveWorkflowRouter {
final case class CommandResult(effect: Workflow.Effect[_])

final case class TransitionalResult(effect: Workflow.Effect.TransitionalEffect[_])
final case class CommandHandlerNotFound(commandName: String) extends RuntimeException {
override def getMessage: String = commandName
}
final case class WorkflowStepNotFound(stepName: String) extends RuntimeException {
override def getMessage: String = stepName
}
Expand Down Expand Up @@ -98,39 +93,27 @@ class ReflectiveWorkflowRouter[S, W <: Workflow[S]](

val workflow = instanceFactory(workflowContext)

val commandEffect =
try {
// if runtime doesn't have a state to provide, we fall back to user's own defined empty state
val decodedState = decodeUserState(userState).getOrElse(workflow.emptyState())
workflow._internalSetup(decodedState, context, timerScheduler)

val methodInvoker = methodInvokerLookup(commandName)

if (serializer.isJson(command) || command.isEmpty) {
// - BytesPayload.empty - there is no real command, and we are calling a method with arity 0
// - BytesPayload with json - we deserialize it and call the method
val deserializedCommand =
CommandSerialization.deserializeComponentClientCommand(methodInvoker.method, command, serializer)
val result = deserializedCommand match {
case None => methodInvoker.invoke(workflow)
case Some(input) => methodInvoker.invokeDirectly(workflow, input)
}
result.asInstanceOf[Workflow.Effect[_]]
} else {
throw new IllegalStateException(
s"Could not find a matching command handler for method [$commandName], content type [${command.contentType}] " +
s"on [${workflow.getClass.getName}]")
}

} catch {
case CommandHandlerNotFound(name) =>
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

was this dead-code?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yeap, I simplified this code before, and nothing throws this exception anymore

throw new WorkflowException(
context.workflowId(),
commandName,
s"No command handler found for command [$name] on [${workflow.getClass.getName}]")
// if runtime doesn't have a state to provide, we fall back to user's own defined empty state
val decodedState = decodeUserState(userState).getOrElse(workflow.emptyState())
workflow._internalSetup(decodedState, context, timerScheduler)

val methodInvoker = methodInvokerLookup(commandName)

if (serializer.isJson(command) || command.isEmpty) {
// - BytesPayload.empty - there is no real command, and we are calling a method with arity 0
// - BytesPayload with json - we deserialize it and call the method
val deserializedCommand =
CommandSerialization.deserializeComponentClientCommand(methodInvoker.method, command, serializer)
val result = deserializedCommand match {
case None => methodInvoker.invoke(workflow)
case Some(input) => methodInvoker.invokeDirectly(workflow, input)
}

CommandResult(commandEffect)
CommandResult(result.asInstanceOf[Workflow.Effect[_]])
} else {
throw new IllegalStateException(
s"Could not find a matching command handler for method [$commandName], content type [${command.contentType}] " +
s"on [${workflow.getClass.getName}]")
}
}

final def handleStep(
Expand Down
17 changes: 14 additions & 3 deletions akka-javasdk/src/main/scala/akka/javasdk/impl/workflow/WorkflowImpl.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,8 @@ import scala.concurrent.Future
import scala.jdk.CollectionConverters.ListHasAsScala
import scala.jdk.DurationConverters.JavaDurationOps
import scala.jdk.OptionConverters.RichOptional
import scala.util.Failure
import scala.util.Success
import scala.util.control.NonFatal

import akka.annotation.InternalApi
Expand Down Expand Up @@ -50,13 +52,16 @@ import akka.runtime.sdk.spi.SpiWorkflow
import akka.runtime.sdk.spi.TimerClient
import io.opentelemetry.api.trace.Span
import io.opentelemetry.api.trace.Tracer
import org.slf4j.Logger
import org.slf4j.LoggerFactory

/**
* INTERNAL API
*/
@InternalApi
class WorkflowImpl[S, W <: Workflow[S]](
workflowId: String,
workflowClass: Class[W],
serializer: JsonSerializer,
componentDescriptor: ComponentDescriptor,
timerClient: TimerClient,
Expand All @@ -65,6 +70,8 @@ class WorkflowImpl[S, W <: Workflow[S]](
instanceFactory: Function[WorkflowContext, W])
extends SpiWorkflow {

private val log: Logger = LoggerFactory.getLogger(workflowClass)

private val context = new WorkflowContextImpl(workflowId)

private val router =
Expand Down Expand Up @@ -228,15 +235,19 @@ class WorkflowImpl[S, W <: Workflow[S]](
new TimerSchedulerImpl(timerClient, context.componentCallMetadata)

try {
router.handleStep(
val handleStep = router.handleStep(
userState,
input = input,
stepName = stepName,
timerScheduler = timerScheduler,
commandContext = context,
executionContext = sdkExecutionContext)
handleStep.onComplete {
case Failure(exception) => log.error(s"Workflow [$workflowId], failed to execute step [$stepName]", exception)
case Success(_) =>
}(sdkExecutionContext)
handleStep
} catch {
case e: WorkflowException => throw e
case NonFatal(ex) =>
throw WorkflowException(s"unexpected exception [${ex.getMessage}] while executing step [$stepName]", Some(ex))
}
Expand All @@ -250,8 +261,8 @@ class WorkflowImpl[S, W <: Workflow[S]](
try {
router.getNextStep(stepName, result.get, userState)
} catch {
case e: WorkflowException => throw e
case NonFatal(ex) =>
log.error(s"Workflow [$workflowId], failed to transition from step [$stepName]", ex)
throw WorkflowException(
s"unexpected exception [${ex.getMessage}] while executing transition for step [$stepName]",
Some(ex))
Expand Down
Loading