class KafkaFilterManager {

    @VisibleForTesting
    public static Optional<Range> filterRangeByDomain(Domain domain)
    {
        Long low = INVALID_KAFKA_RANGE_INDEX;
        Long high = INVALID_KAFKA_RANGE_INDEX;
        if (domain.isSingleValue()) {
            // still return range for single value case like (_partition_offset=XXX or _timestamp=XXX)
            low = (long) domain.getSingleValue();
            high = (long) domain.getSingleValue();
        }
        else {
            ValueSet valueSet = domain.getValues();
            if (valueSet instanceof SortedRangeSet) {
                // still return range for single value case like (_partition_offset in (XXX1,XXX2) or _timestamp in XXX1, XXX2)
                Ranges ranges = ((SortedRangeSet) valueSet).getRanges();
                List<io.prestosql.spi.predicate.Range> rangeList = ranges.getOrderedRanges();
                if (rangeList.stream().allMatch(io.prestosql.spi.predicate.Range::isSingleValue)) {
                    List<Long> values = rangeList.stream()
                            .map(range -> (Long) range.getSingleValue())
                            .collect(toImmutableList());
                    low = Collections.min(values);
                    high = Collections.max(values);
                }
                else {
                    Marker lowMark = ranges.getSpan().getLow();
                    low = getLowByLowMark(lowMark).orElse(low);
                    Marker highMark = ranges.getSpan().getHigh();
                    high = getHighByHighMark(highMark).orElse(high);
                }
            }
        }
        if (high != INVALID_KAFKA_RANGE_INDEX) {
            high = high + 1;
        }
        return Optional.of(new Range(low, high));
    }

    @Inject public  KafkaFilterManager(KafkaConsumerFactory consumerFactory, KafkaAdminFactory adminFactory);

}