Add basic reporting for import batch #4773
Conversation
| val processorsMap = processors.foldLeft(Map.empty[EntityType, EventProcessor[_]]) { (acc, processor) => | ||
| acc + (processor.resourceType -> processor) | ||
| } | ||
| eventStream | ||
| .evalTap { event => | ||
| .evalScan(ImportReport.start) { case(report, event) => |
| processorsMap.get(event.`type`) match { | ||
| case Some(processor) => | ||
| processor.evaluate(event).onError { err => | ||
| processor.evaluate(event) | ||
| .map { status => |
There was a problem hiding this comment.
Not sure if it'll be useful to preserve, but could also do .attempt.map, log the errors and include a failure case with the error info, as well as dropped / success. I've done migrations like this and something like this can help to categorise classes of errors to know where to focus on
There was a problem hiding this comment.
This PR is meant to be just a first step meant to be enriched later when we import more cases and have a better grasp on how to handle error handling / error handling (with something similar to what you propose for example).
| val header = logger.info(s"Type\tSuccess\tDropped") | ||
| val detailedCount = report.progress.foldLeft(header) { case (acc, (entityType, count)) => | ||
| acc >> logger.info(s"$entityType\t${count.success}\t${count.dropped}") | ||
| } | ||
|
|
||
| val aggregatedCount = report.aggregatedCount | ||
| logger.info(s"${aggregatedCount.success} events were imported up to offset ${report.offset} (${aggregatedCount.success} have been dropped).") >> | ||
| detailedCount |
There was a problem hiding this comment.
Could also stick in a Show instance for the report and this becomes .flatTap(report => logger.info(report.show)). (Keeps this cleaner and you could dump it all in a file if needed)
No description provided.