Improve reporting

src/main/kotlin/Service.kt

@@ -24,6 +24,9 @@ import java.util.Properties
 import org.apache.kafka.clients.consumer.ConsumerRecord
 import org.apache.logging.log4j.LogManager
 import org.apache.logging.log4j.Logger
+import org.fcrepo.client.FcrepoOperationFailedException
+import org.memobase.exceptions.MissingMimeTypeException
+import org.memobase.exceptions.SftpClientException
 import org.memobase.fedora.FedoraClient
 import org.memobase.fedora.FedoraClientImpl
 import org.memobase.settings.SettingsLoader
@@ -80,32 +83,57 @@ class Service(fileName: String = "app.yml") : Closeable {
 
     fun processRecords() {
         for (record in consumer.fetchRecords()) {
-            val ingestReport = processRecord(record)
-            producer.sendReport(record.headers(), ingestReport)
+            processRecord(record)
         }
     }
 
-    private fun processRecord(record: ConsumerRecord<String, String>): Report {
+    private fun processRecord(record: ConsumerRecord<String, String>) {
         val ingester = Ingester(
                 settings.sftpSettings,
                 createFedoraClient(settings.appSettings),
                 settings.appSettings.getProperty("$FEDORA_PROPERTIES_PREFIX.externalBaseUrl")
         )
 
-        return try {
+        try {
             ingester.ingest(record.key(), record.value())
-            Report(
+            val report = Report(
                     id = record.key(),
                     status = ReportStatus.success,
                     message = ReportMessages.ingestedRecord(record.key())
             )
+            producer.sendReport(record.headers(), report)
+        } catch (ex: FcrepoOperationFailedException) {
+            log.error("${ex.javaClass.canonicalName}:: ${ex.localizedMessage}", ex)
+            val report = Report(
+                    id = record.key(),
+                    status = ReportStatus.failure,
+                    message = "Fedora Exception: ${ex.localizedMessage}"
+            )
+            producer.sendReport(record.headers(), report)
+        } catch (ex: MissingMimeTypeException) {
+            log.error("${ex.javaClass.canonicalName}:: ${ex.localizedMessage}", ex)
+            val report = Report(
+                    id = record.key(),
+                    status = ReportStatus.failure,
+                    message = "Missing MimeType Exception: ${ex.localizedMessage}"
+                )
+            producer.sendReport(record.headers(), report)
+        } catch (ex: SftpClientException) {
+            log.error("${ex.javaClass.canonicalName}: ${ex.localizedMessage}", ex)
+            val report = Report(
+                    id = record.key(),
+                    status = ReportStatus.failure,
+                    message = "SFTP Exception: ${ex.localizedMessage}"
+                )
+            producer.sendReport(record.headers(), report)
         } catch (ex: Exception) {
-            log.error("Ingestion of record ${record.key()} failed: ${ex.localizedMessage}", ex)
-            Report(
+            log.error("${ex.javaClass.canonicalName}: ${ex.localizedMessage}", ex)
+            val report = Report(
                     id = record.key(),
                     status = ReportStatus.failure,
-                    message = ReportMessages.ingestFailed(record.key())
+                    message = "Unknown Exception: ${ex.localizedMessage}"
                 )
+            producer.sendReport(record.headers(), report)
         }
     }