@Override
    public CompletableFuture<List<Long>> getSplitBoundary(BundleSplitOption bundleSplitOption) {
        NamespaceService service = bundleSplitOption.getService();
        NamespaceBundle bundle = bundleSplitOption.getBundle();
        List<Long> positions = bundleSplitOption.getPositions();
        if (positions == null || positions.size() == 0) {
            throw new IllegalArgumentException("SplitBoundaries can't be empty");
        }
        // sort all positions
        Collections.sort(positions);
        if (force) {
            return getBoundaries(bundle, positions);
        } else {
            return service.getOwnedTopicListForNamespaceBundle(bundle)
                    .thenCompose(topics -> {
                        if (topics == null || topics.size() <= 1) {
                            return CompletableFuture.completedFuture(null);
                        }
                        return getBoundaries(bundle, positions);
                    });
        }
    }
@SuppressWarnings("UnstableApiUsage")
    @Test
    public void testAlgorithmReturnCorrectResult() {
        // -- algorithm
        SpecifiedPositionsBundleSplitAlgorithm algorithm = new SpecifiedPositionsBundleSplitAlgorithm();
        // -- calculate the mock result
        NamespaceService mockNamespaceService = mock(NamespaceService.class);
        NamespaceBundle mockNamespaceBundle = mock(NamespaceBundle.class);
        doReturn(1L).when(mockNamespaceBundle).getLowerEndpoint();
        doReturn(1000L).when(mockNamespaceBundle).getUpperEndpoint();
        doReturn(CompletableFuture.completedFuture(Lists.newArrayList("topic", "topic2")))
                .when(mockNamespaceService)
                .getOwnedTopicListForNamespaceBundle(mockNamespaceBundle);

        List<Long> positions = Arrays.asList(-1L, 0L, 1L, 100L, 200L, 500L, 800L, 1000L, 1100L);
        List<Long> splitPositions = algorithm.getSplitBoundary(
                new BundleSplitOption(mockNamespaceService, mockNamespaceBundle, positions)).join();

        assertEquals(splitPositions.size(), 4);
        assertTrue(splitPositions.contains(100L));
        assertTrue(splitPositions.contains(200L));
        assertTrue(splitPositions.contains(500L));
        assertTrue(splitPositions.contains(800L));
    }