@Override
    public Output run(RunContext runContext) throws Exception {

        URI from = new URI(runContext.render(this.from));

        final PebbleExpressionPredicate predicate = getExpressionPredication(runContext);

        final Path path = runContext.workingDir().createTempFile(".ion");
        long processedItemsTotal = 0L;
        long droppedItemsTotal = 0L;
        try (final BufferedWriter writer = Files.newBufferedWriter(path);
             final BufferedReader reader = newBufferedReader(runContext, from)) {

            String item;
            while ((item = reader.readLine()) != null) {
                IllegalVariableEvaluationException exception = null;
                Boolean match = null;
                try {
                    match = predicate.apply(item);
                } catch (IllegalVariableEvaluationException e) {
                    exception = e;
                }

                FilterType action = this.filterType;

                if (match == null) {
                    switch (errorOrNullBehavior) {
                        case FAIL -> {
                            if (exception != null) {
                                throw exception;
                            } else {
                                throw new IllegalVariableEvaluationException(String.format(
                                    "Expression `%s` return `null` on item `%s`",
                                    filterCondition,
                                    item
                                ));
                            }
                        }
                        case INCLUDE -> action = FilterType.INCLUDE;
                        case EXCLUDE ->  action = FilterType.EXCLUDE;
                    }
                    match = true;
                }

                if (!match) {
                    action = action.reverse();
                }

                switch (action) {
                    case INCLUDE -> {
                        writer.write(item);
                        writer.newLine();
                    }
                    case EXCLUDE -> droppedItemsTotal++;
                }
                processedItemsTotal++;
            }
        }
        URI uri = runContext.storage().putFile(path.toFile());
        return Output.builder()
            .uri(uri)
            .processedItemsTotal(processedItemsTotal)
            .droppedItemsTotal(droppedItemsTotal)
            .build();
    }
@Test
    void shouldThrowExceptionGivenInvalidRecordsForFail() throws Exception {
        // Given
        RunContext runContext = runContextFactory.of();

        FilterItems task = FilterItems
            .builder()
            .from(generateKeyValueFile(TEST_INVALID_ITEMS, runContext).toString())
            .filterCondition(" {{ value % 2 == 0 }}")
            .filterType(FilterItems.FilterType.INCLUDE)
            .errorOrNullBehavior(FilterItems.ErrorOrNullBehavior.FAIL)
            .build();

        // When/Then
        Assertions.assertThrows(IllegalVariableEvaluationException.class, () -> task.run(runContext));
    }