### Question:
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 { 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  KafkaFilterManager(KafkaConsumerFactory consumerFactory, KafkaAdminFactory adminFactory); KafkaFilteringResult getKafkaFilterResult(
            ConnectorSession session,
            KafkaTableHandle kafkaTableHandle,
            List<PartitionInfo> partitionInfos,
            Map<TopicPartition, Long> partitionBeginOffsets,
            Map<TopicPartition, Long> partitionEndOffsets); @VisibleForTesting static Optional<Range> filterRangeByDomain(Domain domain); @VisibleForTesting static Set<Long> filterValuesByDomain(Domain domain, Set<Long> sourceValues);  }### Answer:
@Test public void testFilterValuesByDomain() { Set<Long> source = Set.of(1L, 2L, 3L, 4L, 5L, 6L); Domain testDomain = Domain.singleValue(BIGINT, 1L); assertEquals(KafkaFilterManager.filterValuesByDomain(testDomain, source), Set.of(1L)); testDomain = multipleValues(BIGINT, ImmutableList.of(3L, 8L)); assertEquals(KafkaFilterManager.filterValuesByDomain(testDomain, source), Set.of(3L)); testDomain = Domain.create(SortedRangeSet.copyOf(BIGINT, ImmutableList.of( Range.range(BIGINT, 2L, true, 4L, true))), false); assertEquals(KafkaFilterManager.filterValuesByDomain(testDomain, source), Set.of(2L, 3L, 4L)); }