I'm currently running an Akka streams setup similar to the following:
┌───────────────┐
┌─────────────┐ │┌─────────────┐│
│REST endpoint│──▶│Queue source ││
└─────────────┘ │└──────╷──────┘│
│┌──────▼──────┐│
││ Flow[T] ││
│└──────╷──────┘│
│┌──────▼──────┐│ ┌─────────────┐
││ KafkaSink │├─▶│ Kafka topic │
│└─────────────┘│ └─────────────┘
└───────────────┘
While this working is well, I'd like to have some insight into the production system, i.e. have there been errors and what kind of errors. For example, I've wrapped the KafkaSink into a RestartSink.withBackoff and applied the following attributes to the wrapped sink:
private val decider: Supervision.Decider = {
case x =>
log.error("KafkaSink encountered an error and will stop", x)
Supervision.Stop
}
Flow[...]
.log("KafkaSink")
.to(Producer.plainSink(...))
.withAttributes(ActorAttributes.supervisionStrategy(decider))
.addAttributes(
ActorAttributes.logLevels(
onElement = Logging.DebugLevel,
onFinish = Logging.WarningLevel,
onFailure = Logging.ErrorLevel
)
)
This does provide me with some insight, e.g. I'll get a log message that the downstream has closed as well as the exception that ocurred via the supervisionStrategy that I've added.
This solution, however, feels a little bit like a workaround (e.g. logging the exception the supervision strategy), and it also doesn't provide any insight into the behaviour of the RestartWithBackoffSink. I could, of course, enable DEBUG level logging for that class, but again, that feels like a workaround to be doing in production.
Long story short:
- Are there any obvious shortcomings of the way I'm trying to gain insight into errors happening in my Akka streams
- Is there a better / more idiomatic way of adding logging to Akka streams in production