@Override
    public CompletableFuture<List<Long>> getSplitBoundary(BundleSplitOption bundleSplitOption) {
        NamespaceService service = bundleSplitOption.getService();
        NamespaceBundle bundle = bundleSplitOption.getBundle();
        return service.getOwnedTopicListForNamespaceBundle(bundle).thenCompose(topics -> {
            if (topics == null || topics.size() <= 1) {
                return CompletableFuture.completedFuture(null);
            }
            List<Long> topicNameHashList = new ArrayList<>(topics.size());
            for (String topic : topics) {
                topicNameHashList.add(bundle.getNamespaceBundleFactory().getLongHashCode(topic));
            }
            Collections.sort(topicNameHashList);
            long splitStart = topicNameHashList.get(Math.max((topicNameHashList.size() / 2) - 1, 0));
            long splitEnd = topicNameHashList.get(topicNameHashList.size() / 2);
            long splitMiddle = splitStart + (splitEnd - splitStart) / 2;
            return CompletableFuture.completedFuture(Collections.singletonList(splitMiddle));
        });
    }
@SuppressWarnings("UnstableApiUsage")
    @Test
    public void testAlgorithmReturnCorrectResult() {
        // -- algorithm
        TopicCountEquallyDivideBundleSplitAlgorithm algorithm = new TopicCountEquallyDivideBundleSplitAlgorithm();
        List<String> mockTopics = Lists.newArrayList("a", "b", "c");
        // -- calculate the mock result
        NamespaceService namespaceServiceForMockResult = mock(NamespaceService.class);
        NamespaceBundle namespaceBundleForMockResult = mock(NamespaceBundle.class);
        doReturn(CompletableFuture.completedFuture(mockTopics))
                .when(namespaceServiceForMockResult).getOwnedTopicListForNamespaceBundle(namespaceBundleForMockResult);
        List<Long> hashList = new ArrayList<>();
        NamespaceBundleFactory namespaceBundleFactoryForMockResult = mock(NamespaceBundleFactory.class);
        mockTopics.forEach((topic) -> {
            long hashValue = Hashing.crc32().hashString(topic, StandardCharsets.UTF_8).padToLong();
            doReturn(namespaceBundleFactoryForMockResult)
                    .when(namespaceBundleForMockResult).getNamespaceBundleFactory();
            doReturn(hashValue)
                    .when(namespaceBundleFactoryForMockResult).getLongHashCode(topic);
            hashList.add(hashValue);
        });
        Collections.sort(hashList);
        long splitStart = hashList.get(Math.max((hashList.size() / 2) - 1, 0));
        long splitEnd = hashList.get(hashList.size() / 2);
        long splitMiddleForMockResult = splitStart + (splitEnd - splitStart) / 2;
        // -- do test
        NamespaceService mockNamespaceService = mock(NamespaceService.class);
        NamespaceBundle mockNamespaceBundle = mock(NamespaceBundle.class);
        doReturn(CompletableFuture.completedFuture(mockTopics))
                .when(mockNamespaceService).getOwnedTopicListForNamespaceBundle(mockNamespaceBundle);
        NamespaceBundleFactory mockNamespaceBundleFactory = mock(NamespaceBundleFactory.class);
        mockTopics.forEach((topic) -> {
            doReturn(mockNamespaceBundleFactory)
                    .when(mockNamespaceBundle).getNamespaceBundleFactory();
            long hashValue = Hashing.crc32().hashString(topic, StandardCharsets.UTF_8).padToLong();
            doReturn(hashValue)
                    .when(mockNamespaceBundleFactory).getLongHashCode(topic);
        });
        assertEquals((long) algorithm.getSplitBoundary(new BundleSplitOption(mockNamespaceService, mockNamespaceBundle, null)).join().get(0),
                splitMiddleForMockResult);
    }