feat(api): add access to error stacktrace in filter chain

Author: fhussonnois

connect-file-pulse-api/src/main/java/io/streamthoughts/kafka/connect/filepulse/filter/DefaultRecordFilterPipeline.java

@@ -195,7 +195,7 @@ public List<FileRecord<TypedStruct>> apply(final FilterContext context,
 
                     if (filter.onFailure() != null) {
                         final FilterContext errorContext = FilterContextBuilder.newBuilder(context)
-                                .withException(new FilterError(e.getLocalizedMessage(), filter.label()))
+                                .withError(FilterError.of(e, filter.label()))
                                 .build();
                         filtered.addAll(filter.onFailure().apply(errorContext, record, hasNext));
                     } else {

connect-file-pulse-api/src/main/java/io/streamthoughts/kafka/connect/filepulse/filter/FilterContextBuilder.java

@@ -30,7 +30,7 @@
  */
 public class FilterContextBuilder {
 
-    private FilterError exception;
+    private FilterError error;
     private FileObjectMeta metadata;
     private FileRecordOffset offset;
     private String topic;
@@ -65,7 +65,7 @@ static FilterContextBuilder newBuilder(final FilterContext original) {
                 .withOffset(original.offset())
                 .withVariables(original.variables())
                 .withHeaders(original.headers())
-                .withException(original.error());
+                .withError(original.error());
     }
 
     FilterContextBuilder withVariables(final Map<String, Object> variables) {
@@ -108,8 +108,8 @@ FilterContextBuilder withOffset(final FileRecordOffset offset) {
         return this;
     }
 
-    FilterContextBuilder withException(final FilterError exception) {
-        this.exception = exception;
+    FilterContextBuilder withError(final FilterError exception) {
+        this.error = exception;
         return this;
     }
 
@@ -125,7 +125,7 @@ public FilterContext build() {
                 timestamp,
                 key,
                 headers,
-                exception,
+                error,
                 variables);
     }
 }

connect-file-pulse-api/src/main/java/io/streamthoughts/kafka/connect/filepulse/filter/FilterError.java

@@ -18,40 +18,96 @@
  */
 package io.streamthoughts.kafka.connect.filepulse.filter;
 
+import java.io.PrintWriter;
+import java.io.StringWriter;
 import java.util.Objects;
 
 public class FilterError {
 
-    private final String message;
+    private final String exceptionMessage;
+
+    private final String exceptionClassName;
+
+    private final String exceptionStacktrace;
 
     private final String filter;
 
     /**
-     * Creates a new {@link FilterError} instance.
-     *
-     * @param error     the error message message.
-     * @param filter    the filter name that failed.
+     * Helper method to create a new {@link FilterError} object from a given {@link Throwable}.
      */
-    FilterError(final String error, final String filter) {
-        Objects.requireNonNull(error, "The message message");
-        Objects.requireNonNull(filter, "The filter message");
-        this.message = error;
-        this.filter = filter;
+    public static FilterError of(final Throwable throwable, final String filter) {
+        final Throwable cause;
+        if (throwable instanceof FilterException && throwable.getCause() != null) {
+            cause = throwable.getCause();
+        } else {
+            cause = throwable;
+        }
+        return new FilterError(cause.getMessage(), cause.getClass().getName(), getStacktrace(cause), filter);
     }
 
-    public String message() {
-        return message;
+    /**
+     * Creates a new {@link FilterError} instance.
+     *
+     * @param message       the exception message.
+     * @param classname     the exception className.
+     * @param stacktrace    the exception stacktrace.
+     * @param filter        the failed filter name.
+     */
+    public FilterError(final String message,
+                       final String classname,
+                       final String stacktrace,
+                       final String filter) {
+        this.exceptionMessage = Objects.requireNonNull(message, "'message' should not be null");
+        this.exceptionClassName = Objects.requireNonNull(classname, "'classname' should not be null");
+        this.exceptionStacktrace = Objects.requireNonNull(stacktrace, "'stacktrace' should not be null");
+        this.filter = Objects.requireNonNull(filter, "'filter' should not be null");
     }
 
     public String filter() {
         return filter;
     }
 
+    public String exceptionMessage() {
+        return exceptionMessage;
+    }
+
+    public String exceptionStacktrace() {
+        return exceptionStacktrace;
+    }
+
+    public String getExceptionClassName() {
+        return exceptionClassName;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        FilterError that = (FilterError) o;
+        return Objects.equals(exceptionMessage, that.exceptionMessage) &&
+               Objects.equals(exceptionClassName, that.exceptionClassName) &&
+               Objects.equals(exceptionStacktrace, that.exceptionStacktrace) &&
+               Objects.equals(filter, that.filter);
+    }
+
+    /**
+     * {@inheritDoc}
+     */
     @Override
-    public String toString() {
-        return "[" +
-                "message=" + message +
-                ", filter=" + filter +
-                ']';
+    public int hashCode() {
+        return Objects.hash(exceptionMessage, exceptionClassName, exceptionStacktrace, filter);
+    }
+
+    /**
+     * @return the stacktrace representation for the given {@link Throwable}.
+     */
+    private static String getStacktrace(final Throwable throwable) {
+        final StringWriter sw = new StringWriter();
+        final PrintWriter pw = new PrintWriter(sw, true);
+        throwable.printStackTrace(pw);
+        return sw.getBuffer().toString();
     }
 }