[Broker] Add timeout to closing CoordinationServiceImpl
Fixes
Also close the executor that wasn't closed
package org.apache.pulsar.metadata.coordination.impl;
import io.netty.util.concurrent.DefaultThreadFactory;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.EnumSet;
import java.util.List;
@@ -29,6 +30,7 @@ import java.util.concurrent.CompletionException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import org.apache.bookkeeper.common.concurrent.FutureUtils;
import org.apache.pulsar.metadata.api.MetadataSerde;
@@ -43,6 +45,7 @@ import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
@SuppressWarnings("unchecked")
public class CoordinationServiceImpl implements CoordinationService {
+ private static final Duration CLOSE_TIMEOUT = Duration.ofSeconds(10);
private final MetadataStoreExtended store;
private final Map<Object, LockManager<?>> lockManagers = new ConcurrentHashMap<>();
@@ -69,9 +72,11 @@ public class CoordinationServiceImpl implements CoordinationService {
futures.add(lm.asyncClose());
}
- FutureUtils.collect(futures).join();
+ FutureUtils.collect(futures).get(CLOSE_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
} catch (CompletionException ce) {
throw MetadataStoreException.unwrap(ce);
+ } finally {
+ executor.shutdownNow();
}
}
