Destroy visitor session in other thread to avoid deadlock, and 1000 queue size
@@ -40,6 +40,7 @@ import java.util.StringJoiner;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
@@ -85,6 +86,7 @@ public class DocumentOperationExecutorImpl implements DocumentOperationExecutor
private final DelayQueue throttled;
private final DelayQueue timeouts;
private final Map<VisitorControlHandler, VisitorSession> visits = new ConcurrentHashMap<>();
+ private final Executor visitSessionShutdownExecutor = Executors.newSingleThreadExecutor();
public DocumentOperationExecutorImpl(ClusterListConfig clustersConfig, AllClustersBucketSpacesConfig bucketsConfig,
DocumentOperationExecutorConfig executorConfig, DocumentAccess access, Clock clock) {
@@ -212,12 +214,18 @@ public class DocumentOperationExecutorImpl implements DocumentOperationExecutor
context.error(ERROR, message != null ? message : "Visiting failed");
}
done.set(true); // This may be reached before dispatching thread is done putting us in the map.
- visits.computeIfPresent(this, (__, session) -> { session.destroy(); return null; });
+ visits.computeIfPresent(this, (__, session) -> {
+ visitSessionShutdownExecutor.execute(() -> session.destroy());
+ return null;
+ });
}
});
visits.put(parameters.getControlHandler(), access.createVisitorSession(parameters));
if (done.get())
- visits.computeIfPresent(parameters.getControlHandler(), (__, session) -> { session.destroy(); return null; });
+ visits.computeIfPresent(parameters.getControlHandler(), (__, session) -> {
+ visitSessionShutdownExecutor.execute(() -> session.destroy());
+ return null;
+ });
}
catch (IllegalArgumentException | ParseException e) {
context.error(BAD_REQUEST, Exceptions.toMessageString(e));
