FilterableTableSource does not use filters on job run

public class CustomTableSource implements BatchTableSource<Model>, FilterableTableSource<Model> {

    private static final Logger LOG = LoggerFactory.getLogger(CustomTableSource.class);

    private final Filter[] filters;

    private final FilterConverter converter = new FilterConverter();

    public CustomTableSource() {
        this(null);
    }

    private CustomTableSource(Filter[] filters) {
        this.filters = filters;
    }

    @Override
    public DataSet<Model> getDataSet(ExecutionEnvironment execEnv) {
        if (filters == null) {
           LOG.info("==== No filters defined ====");
        } else {
            LOG.info("==== Found filters ====");
            for (Filter filter : filters) {
                LOG.info("FILTER: {}", filter);
            }
        }

        return execEnv.fromCollection(allModels());
    }

    @Override
    public TableSource<Model> applyPredicate(List<Expression> predicates) {
        LOG.info("Applying predicates");

        List<Filter> acceptedFilters = new ArrayList<>();
        for (final Expression predicate : predicates) {
            converter.convert(predicate).ifPresent(acceptedFilters::add);
        }

        return new CustomTableSource(acceptedFilters.toArray(new Filter[0]));
    }

    @Override
    public boolean isFilterPushedDown() {
        return filters != null;
    }

    @Override
    public TypeInformation<Model> getReturnType() {
        return TypeInformation.of(Model.class);
    }

    @Override
    public TableSchema getTableSchema() {
        return TableSchema.fromTypeInfo(getReturnType());
    }

    private List<Model> allModels() {
        List<Model> models = new ArrayList<>();

        models.add(new Model(1, 2, 3, 4));
        models.add(new Model(10, 11, 12, 13));
        models.add(new Model(20, 21, 22, 23));

        return models;
    }
}