### Question:
StateTransferManager { Stream<TransferBatchRequest> rangeToBatchRequestStream(TransferSegmentRangeSingle range) { ImmutableList<Long> unknownAddressesInRange = range.getUnknownAddressesInRange(); Optional<ImmutableList<String>> availableServers = range.getAvailableServers(); return Lists.partition(unknownAddressesInRange, batchSize).stream() .map(partition -> TransferBatchRequest .builder() .addresses(partition) .destinationNodes(availableServers) .build()); }  ImmutableList<TransferSegment> handleTransfer(List<TransferSegmentRange> beforeTransferRanges);  }### Answer:
@Test void testRangeToBatchRequestStreamEmpty() { StateTransferManager defaultInstance = getDefaultInstance(); TransferSegmentRangeSingle singleRange = (TransferSegmentRangeSingle) getSingleRange(0L, 1000L, StateTransferType.PROTOCOL_READ, false, Optional.empty(), TransferSegmentStatus.builder().build()); Stream<TransferBatchRequest> transferBatchRequestStream = defaultInstance.rangeToBatchRequestStream(singleRange); ImmutableList<TransferBatchRequest> collect = transferBatchRequestStream.collect(ImmutableList.toImmutableList()); assertThat(collect).isEmpty(); }
@Test void testRangeToBatchRequestStreamNoServers() { StateTransferManager defaultInstance = getDefaultInstance(); TransferSegmentRangeSingle singleRange = (TransferSegmentRangeSingle) getSingleRange(0L, 19L, StateTransferType.PROTOCOL_READ, false, Optional.empty(), TransferSegmentStatus.builder().build()); ImmutableList<Long> unknown = LongStream.range(0L, 20L).boxed().collect(ImmutableList.toImmutableList()); singleRange = singleRange.toBuilder().unknownAddressesInRange(unknown).build(); Stream<TransferBatchRequest> transferBatchRequestStream = defaultInstance.rangeToBatchRequestStream(singleRange); ImmutableList<TransferBatchRequest> collect = transferBatchRequestStream.collect(ImmutableList.toImmutableList()); ImmutableList<TransferBatchRequest> expected = ImmutableList.of(TransferBatchRequest .builder().addresses(LongStream.range(0L, 10L) .boxed().collect(ImmutableList.toImmutableList())).build(), TransferBatchRequest.builder().addresses(LongStream.range(10L, 20L) .boxed().collect(ImmutableList.toImmutableList())).build()); assertThat(collect).isEqualTo(expected); }
@Test void testRangeToBatchRequestStreamWithServers() { StateTransferManager defaultInstance = getDefaultInstance(); Optional<ImmutableList<String>> nodes = Optional.of(ImmutableList.of("server1", "server2")); TransferSegmentRangeSingle singleRange = (TransferSegmentRangeSingle) getSingleRange(0L, 19L, CONSISTENT_READ, false, nodes, TransferSegmentStatus.builder().build()); ImmutableList<Long> unknown = LongStream.range(0L, 20L).boxed().collect(ImmutableList.toImmutableList()); singleRange = singleRange.toBuilder().unknownAddressesInRange(unknown).build(); Stream<TransferBatchRequest> transferBatchRequestStream = defaultInstance.rangeToBatchRequestStream(singleRange); ImmutableList<TransferBatchRequest> collect = transferBatchRequestStream.collect(ImmutableList.toImmutableList()); ImmutableList<TransferBatchRequest> expected = ImmutableList.of(TransferBatchRequest .builder().addresses(LongStream.range(0L, 10L) .boxed().collect(ImmutableList.toImmutableList())).destinationNodes(nodes).build(), TransferBatchRequest.builder().addresses(LongStream.range(10L, 20L) .boxed().collect(ImmutableList.toImmutableList())).destinationNodes(nodes).build()); assertThat(collect).isEqualTo(expected); }