### Question:
KafkaFilterManager { @VisibleForTesting public static Optional<Range> filterRangeByDomain(Domain domain) { Long low = INVALID_KAFKA_RANGE_INDEX; Long high = INVALID_KAFKA_RANGE_INDEX; if (domain.isSingleValue()) { low = (long) domain.getSingleValue(); high = (long) domain.getSingleValue(); } 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)) { 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  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 testFilterRangeByDomain() { Domain testDomain = Domain.singleValue(BIGINT, 1L); assertTrue(KafkaFilterManager.filterRangeByDomain(testDomain).isPresent()); assertEquals(KafkaFilterManager.filterRangeByDomain(testDomain).get().getBegin(), 1L); assertEquals(KafkaFilterManager.filterRangeByDomain(testDomain).get().getEnd(), 2L); testDomain = multipleValues(BIGINT, ImmutableList.of(3L, 8L)); assertTrue(KafkaFilterManager.filterRangeByDomain(testDomain).isPresent()); assertEquals(KafkaFilterManager.filterRangeByDomain(testDomain).get().getBegin(), 3L); assertEquals(KafkaFilterManager.filterRangeByDomain(testDomain).get().getEnd(), 9L); testDomain = Domain.create(SortedRangeSet.copyOf(BIGINT, ImmutableList.of( Range.range(BIGINT, 2L, true, 4L, true))), false); assertTrue(KafkaFilterManager.filterRangeByDomain(testDomain).isPresent()); assertEquals(KafkaFilterManager.filterRangeByDomain(testDomain).get().getBegin(), 2L); assertEquals(KafkaFilterManager.filterRangeByDomain(testDomain).get().getEnd(), 5L); }