resolved comments

apache · Sep 20, 2024 · bee4bd5 · bee4bd5
1 parent 44c3604
commit bee4bd5
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 5 deletions.
diff --git a/.../apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateTableViewSyncer.java b/.../apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateTableViewSyncer.java
@@ -89,8 +89,7 @@ private CompletableFuture<Void> syncToMetadataStore(String key, ServiceUnitState
         return metadataStoreTableView.put(key, data);
     }
 
-    private CompletableFuture<Void> dummy(String key, ServiceUnitStateData data) {
-        return CompletableFuture.completedFuture(null);
+    private void dummy(String key, ServiceUnitStateData data) {
     }
 
     private void syncExistingItems()

diff --git a/...a/src/main/java/org/apache/pulsar/metadata/tableview/impl/MetadataStoreTableViewImpl.java b/...a/src/main/java/org/apache/pulsar/metadata/tableview/impl/MetadataStoreTableViewImpl.java
@@ -234,14 +234,15 @@ private CompletableFuture<Void> handleExisting(String path) {
     }
 
     private void fill() throws MetadataStoreException {
-        long started = System.currentTimeMillis();
+        final var deadline = System.currentTimeMillis() + FILL_TIMEOUT_IN_MILLIS;
         log.info("{} start filling existing items under the pathPrefix:{}", name, pathPrefix);
         ConcurrentLinkedDeque<String> q = new ConcurrentLinkedDeque<>();
         List<CompletableFuture<Void>> futures = new ArrayList<>();
         q.add(pathPrefix);
         LongAdder count = new LongAdder();
         while (!q.isEmpty()) {
-            if (System.currentTimeMillis() - started > FILL_TIMEOUT_IN_MILLIS) {
+            var now = System.currentTimeMillis();
+            if (now >= deadline) {
                 String err = name + " failed to fill existing items in "
                         + TimeUnit.MILLISECONDS.toSeconds(FILL_TIMEOUT_IN_MILLIS) + " secs. Filled count:"
                         + count.sum();
@@ -266,7 +267,9 @@ private void fill() throws MetadataStoreException {
                         }));
             }
             try {
-                FutureUtil.waitForAll(futures).get(timeoutInMillis, TimeUnit.MILLISECONDS);
+                FutureUtil.waitForAll(futures).get(
+                        Math.min(timeoutInMillis, deadline - now),
+                        TimeUnit.MILLISECONDS);
             } catch (Throwable e) {
                 Throwable c = FutureUtil.unwrapCompletionException(e);
                 log.error("{} failed to fill existing items", name, c);