class KafkaFilterManager {

    @VisibleForTesting
    public static Set<Long> filterValuesByDomain(Domain domain, Set<Long> sourceValues)
    {
        requireNonNull(sourceValues, "sourceValues is none");
        if (domain.isSingleValue()) {
            long singleValue = (long) domain.getSingleValue();
            return sourceValues.stream().filter(sourceValue -> sourceValue == singleValue).collect(toImmutableSet());
        }
        else {
            ValueSet valueSet = domain.getValues();
            if (valueSet instanceof SortedRangeSet) {
                Ranges ranges = ((SortedRangeSet) valueSet).getRanges();
                List<io.prestosql.spi.predicate.Range> rangeList = ranges.getOrderedRanges();
                if (rangeList.stream().allMatch(io.prestosql.spi.predicate.Range::isSingleValue)) {
                    return rangeList.stream()
                            .map(range -> (Long) range.getSingleValue())
                            .filter(sourceValues::contains)
                            .collect(toImmutableSet());
                }
                else {
                    // still return values for range case like (_partition_id > 1)
                    long low = 0;
                    long high = Long.MAX_VALUE;

                    Marker lowMark = ranges.getSpan().getLow();
                    low = maxLow(low, lowMark);
                    Marker highMark = ranges.getSpan().getHigh();
                    high = minHigh(high, highMark);
                    final long finalLow = low;
                    final long finalHigh = high;
                    return sourceValues.stream()
                            .filter(item -> item >= finalLow && item <= finalHigh)
                            .collect(toImmutableSet());
                }
            }
        }
        return sourceValues;
    }

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

}