@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);
                    });
        }
    }
@Test
    public void testEmptyTopicWithForce() {
        SpecifiedPositionsBundleSplitAlgorithm algorithm = new SpecifiedPositionsBundleSplitAlgorithm(true);
        NamespaceService mockNamespaceService = mock(NamespaceService.class);
        NamespaceBundle mockNamespaceBundle = mock(NamespaceBundle.class);
        doReturn(1L).when(mockNamespaceBundle).getLowerEndpoint();
        doReturn(1000L).when(mockNamespaceBundle).getUpperEndpoint();
        doReturn(CompletableFuture.completedFuture(List.of()))
                .when(mockNamespaceService).getOwnedTopicListForNamespaceBundle(mockNamespaceBundle);
        List<Long> splitPositions =
                algorithm.getSplitBoundary(new BundleSplitOption(mockNamespaceService, mockNamespaceBundle,
                        Arrays.asList(1L, 2L))).join();
        assertEquals(splitPositions, Arrays.asList(2L));
    }