Update S3 processor to emit errors

LangStream · Jan 29, 2024 · 32a8ac5 · 32a8ac5
1 parent e5e229a
commit 32a8ac5
Showing 1 changed file with 2 additions and 1 deletion.
diff --git a/langstream-agents/langstream-agent-s3/src/main/java/ai/langstream/agents/s3/S3Processor.java b/langstream-agents/langstream-agent-s3/src/main/java/ai/langstream/agents/s3/S3Processor.java
@@ -113,8 +113,9 @@ private void processRecord(Record record, RecordSink recordSink) {
             log.debug("Processed record value: {}", processedRecord.value());
             // Send the processed record to the record sink
             recordSink.emit(new SourceRecordAndResult(record, List.of(processedRecord), null));
-        } catch (Exception e) {
+        } catch (Throwable e) {
             log.error("Error processing record: {}", e.getMessage());
+            recordSink.emit(new SourceRecordAndResult(record, null, e));
         }
     }