diff --git a/src/main/java/com/amazon/opendistroforelasticsearch/ad/transport/AnomalyResultTransportAction.java b/src/main/java/com/amazon/opendistroforelasticsearch/ad/transport/AnomalyResultTransportAction.java
index b4cdfe2f..a2f8abd1 100644
--- a/src/main/java/com/amazon/opendistroforelasticsearch/ad/transport/AnomalyResultTransportAction.java
+++ b/src/main/java/com/amazon/opendistroforelasticsearch/ad/transport/AnomalyResultTransportAction.java
@@ -24,7 +24,7 @@
 import java.util.Optional;
 import java.util.concurrent.Callable;
 import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 import java.time.Instant;
 
@@ -62,7 +62,6 @@
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionListenerResponseHandler;
 import org.elasticsearch.action.ActionRequest;
-import org.elasticsearch.action.LatchedActionListener;
 import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
 import org.elasticsearch.action.bulk.BackoffPolicy;
 import org.elasticsearch.action.index.IndexRequest;
@@ -114,7 +113,6 @@ public class AnomalyResultTransportAction extends HandledTransportAction<ActionR
     private final ADStateManager stateManager;
     private final ColdStartRunner globalRunner;
     private final AnomalyDetectionIndices anomalyDetectionIndices;
-    private volatile TimeValue requestTimeout;
     private final Client client;
     private final FeatureManager featureManager;
     private final ModelManager modelManager;
@@ -151,12 +149,14 @@ public AnomalyResultTransportAction(
         this.stateManager = manager;
         this.globalRunner = eventExecutor;
         this.anomalyDetectionIndices = anomalyDetectionIndices;
-        this.requestTimeout = AnomalyDetectorSettings.REQUEST_TIMEOUT.get(settings);
         this.featureManager = featureManager;
         this.modelManager = modelManager;
         this.hashRing = hashRing;
-        this.requestTimeout = AnomalyDetectorSettings.REQUEST_TIMEOUT.get(settings);
-        this.option = TransportRequestOptions.builder().withType(TransportRequestOptions.Type.REG).withTimeout(requestTimeout).build();
+        this.option = TransportRequestOptions
+            .builder()
+            .withType(TransportRequestOptions.Type.REG)
+            .withTimeout(AnomalyDetectorSettings.REQUEST_TIMEOUT.get(settings))
+            .build();
         this.clusterService = clusterService;
         this.indexNameExpressionResolver = indexNameExpressionResolver;
         this.threadPool = threadPool;
@@ -262,9 +262,7 @@ protected void doExecute(Task task, ActionRequest actionRequest, ActionListener<
                 return;
             }
 
-            String thresholdNodeId = thresholdNode.get().getId();
-
-            if (!shouldStart(listener, adID, detector.get(), thresholdNodeId, thresholdModelID)) {
+            if (!shouldStart(listener, adID, detector.get(), thresholdNode.get().getId(), thresholdModelID)) {
                 return;
             }
 
@@ -302,11 +300,12 @@ protected void doExecute(Task task, ActionRequest actionRequest, ActionListener<
             // detector changes or the cluster size changes.
             int rcfPartitionNum = stateManager.getPartitionNumber(adID);
 
-            final CountDownLatch rcfLatch = createCountDownLatch(rcfPartitionNum);
             List<RCFResultResponse> rcfResults = new ArrayList<>();
 
             final AtomicReference<AnomalyDetectionException> failure = new AtomicReference<AnomalyDetectionException>();
 
+            final AtomicInteger responseCount = new AtomicInteger();
+
             for (int i = 0; i < rcfPartitionNum; i++) {
                 String rcfModelID = modelManager.getRcfModelId(adID, i);
 
@@ -321,10 +320,24 @@ protected void doExecute(Task task, ActionRequest actionRequest, ActionListener<
                 }
 
                 LOG.info("Sending RCF request to {} for model {}", rcfNodeId, rcfModelID);
-                LatchedActionListener<RCFResultResponse> rcfListener = new LatchedActionListener<>(
-                    new RCFActionListener(rcfResults, rcfModelID.toString(), failure, rcfNodeId),
-                    rcfLatch
+
+                RCFActionListener rcfListener = new RCFActionListener(
+                    rcfResults,
+                    rcfModelID.toString(),
+                    failure,
+                    rcfNodeId,
+                    detector,
+                    listener,
+                    thresholdModelID,
+                    thresholdNode,
+                    featureInResponse,
+                    startTime,
+                    endTime,
+                    rcfPartitionNum,
+                    responseCount,
+                    adID
                 );
+
                 transportService
                     .sendRequest(
                         rcfNode.get(),
@@ -334,87 +347,9 @@ protected void doExecute(Task task, ActionRequest actionRequest, ActionListener<
                         new ActionListenerResponseHandler<>(rcfListener, RCFResultResponse::new)
                     );
             }
-
-            // wait a bit longer than transport timeout
-            long latchWaitSecs = Math.round(requestTimeout.getSeconds() * 1.25d);
-
-            try {
-                LOG.debug("Wait for RCF results...");
-                rcfLatch.await(latchWaitSecs, TimeUnit.SECONDS);
-            } catch (InterruptedException e) {
-                listener.onFailure(new InternalFailure(adID, CommonErrorMessages.WAIT_ERR_MSG, e));
-                return;
-            }
-
-            if (coldStartIfNoModel(failure, detector.get()) || rcfResults.isEmpty()) {
-                listener.onFailure(new InternalFailure(adID, NO_MODEL_ERR_MSG));
-                return;
-            }
-
-            CombinedRcfResult combinedResult = getCombinedResult(rcfResults);
-            double combinedScore = combinedResult.getScore();
-
-            final CountDownLatch thresholdLatch = createCountDownLatch(1);
-
-            final AtomicReference<AnomalyResultResponse> anomalyResultResponse = new AtomicReference<>();
-
-            LOG.info("Sending threshold request to {} for model {}", thresholdNodeId, thresholdModelID);
-            LatchedActionListener<ThresholdResultResponse> thresholdListener = new LatchedActionListener<>(
-                new ThresholdActionListener(anomalyResultResponse, featureInResponse, thresholdModelID, failure, thresholdNodeId),
-                thresholdLatch
-            );
-            transportService
-                .sendRequest(
-                    thresholdNode.get(),
-                    ThresholdResultAction.NAME,
-                    new ThresholdResultRequest(adID, thresholdModelID, combinedScore),
-                    option,
-                    new ActionListenerResponseHandler<>(thresholdListener, ThresholdResultResponse::new)
-                );
-
-            try {
-                LOG.debug("Wait for threshold results...");
-                thresholdLatch.await(latchWaitSecs, TimeUnit.SECONDS);
-            } catch (InterruptedException e) {
-                listener.onFailure(new InternalFailure(adID, WAIT_FOR_THRESHOLD_ERR_MSG, e));
-                return;
-            }
-
-            if (coldStartIfNoModel(failure, detector.get())) {
-                listener.onFailure(new InternalFailure(adID, NO_MODEL_ERR_MSG));
-                return;
-            }
-
-            if (anomalyResultResponse.get() != null) {
-                AnomalyResultResponse response = anomalyResultResponse.get();
-                double confidence = response.getConfidence() * combinedResult.getConfidence();
-                response = new AnomalyResultResponse(response.getAnomalyGrade(), confidence, response.getFeatures());
-                listener.onResponse(response);
-                indexAnomalyResult(
-                    new AnomalyResult(
-                        adID,
-                        Double.valueOf(combinedScore),
-                        Double.valueOf(response.getAnomalyGrade()),
-                        Double.valueOf(confidence),
-                        featureInResponse,
-                        Instant.ofEpochMilli(startTime),
-                        Instant.ofEpochMilli(endTime)
-                    )
-                );
-            } else if (failure.get() != null) {
-                listener.onFailure(failure.get());
-            } else {
-                listener.onFailure(new InternalFailure(adID, "Unexpected exception"));
-            }
-        } catch (ClientException clientException) {
-            listener.onFailure(clientException);
-        } catch (AnomalyDetectionException adEx) {
-            listener.onFailure(new InternalFailure(adEx));
-        } catch (Exception throwable) {
-            Throwable cause = ExceptionsHelper.unwrapCause(throwable);
-            listener.onFailure(new InternalFailure(adID, cause));
+        } catch (Exception ex) {
+            handleExecuteException(ex, listener, adID);
         }
-
     }
 
     /**
@@ -637,37 +572,134 @@ void saveDetectorResult(IndexRequest indexRequest, String context, Iterator<Time
             );
     }
 
+    void handleExecuteException(Exception ex, ActionListener<AnomalyResultResponse> listener, String adID) {
+        if (ex instanceof ClientException) {
+            listener.onFailure(ex);
+        } else if (ex instanceof AnomalyDetectionException) {
+            listener.onFailure(new InternalFailure((AnomalyDetectionException) ex));
+        } else {
+            Throwable cause = ExceptionsHelper.unwrapCause(ex);
+            listener.onFailure(new InternalFailure(adID, cause));
+        }
+    }
+
     class RCFActionListener implements ActionListener<RCFResultResponse> {
         private List<RCFResultResponse> rcfResults;
         private String modelID;
         private AtomicReference<AnomalyDetectionException> failure;
-        private String nodeID;
+        private String rcfNodeID;
+        private Optional<AnomalyDetector> detector;
+        private ActionListener<AnomalyResultResponse> listener;
+        private String thresholdModelID;
+        private Optional<DiscoveryNode> thresholdNode;
+        private List<FeatureData> featureInResponse;
+        private long startTime;
+        private long endTime;
+        private int nodeCount;
+        private final AtomicInteger responseCount;
+        private final String adID;
 
         RCFActionListener(
             List<RCFResultResponse> rcfResults,
             String modelID,
             AtomicReference<AnomalyDetectionException> failure,
-            String nodeID
+            String rcfNodeID,
+            Optional<AnomalyDetector> detector,
+            ActionListener<AnomalyResultResponse> listener,
+            String thresholdModelID,
+            Optional<DiscoveryNode> thresholdNode,
+            List<FeatureData> features,
+            long startTime,
+            long endTime,
+            int nodeCount,
+            AtomicInteger responseCount,
+            String adID
         ) {
             this.rcfResults = rcfResults;
             this.modelID = modelID;
+            this.rcfNodeID = rcfNodeID;
+            this.detector = detector;
+            this.listener = listener;
+            this.thresholdNode = thresholdNode;
+            this.thresholdModelID = thresholdModelID;
+            this.featureInResponse = features;
             this.failure = failure;
-            this.nodeID = nodeID;
+            this.startTime = startTime;
+            this.endTime = endTime;
+            this.nodeCount = nodeCount;
+            this.responseCount = responseCount;
+            this.adID = adID;
         }
 
         @Override
         public void onResponse(RCFResultResponse response) {
-            stateManager.resetBackpressureCounter(nodeID);
-            if (response != null) {
-                rcfResults.add(response);
-            } else {
-                LOG.warn(NULL_RESPONSE + " {} for {}", modelID, nodeID);
+            try {
+                stateManager.resetBackpressureCounter(rcfNodeID);
+                if (response != null) {
+                    rcfResults.add(response);
+                } else {
+                    LOG.warn(NULL_RESPONSE + " {} for {}", modelID, rcfNodeID);
+                }
+            } catch (Exception ex) {
+                LOG.error("Unexpected exception: {} for {}", ex, adID);
+            } finally {
+                if (nodeCount == responseCount.incrementAndGet()) {
+                    handleRCFResults();
+                }
             }
         }
 
         @Override
         public void onFailure(Exception e) {
-            handlePredictionFailure(e, modelID, nodeID, failure);
+            try {
+                handlePredictionFailure(e, modelID, rcfNodeID, failure);
+            } catch (Exception ex) {
+                LOG.error("Unexpected exception: {} for {}", ex, adID);
+            } finally {
+                if (nodeCount == responseCount.incrementAndGet()) {
+                    handleRCFResults();
+                }
+            }
+        }
+
+        private void handleRCFResults() {
+            try {
+                if (coldStartIfNoModel(failure, detector.get()) || rcfResults.isEmpty()) {
+                    listener.onFailure(new InternalFailure(adID, NO_MODEL_ERR_MSG));
+                    return;
+                }
+
+                CombinedRcfResult combinedResult = getCombinedResult(rcfResults);
+                double combinedScore = combinedResult.getScore();
+
+                final AtomicReference<AnomalyResultResponse> anomalyResultResponse = new AtomicReference<>();
+
+                String thresholdNodeId = thresholdNode.get().getId();
+                LOG.info("Sending threshold request to {} for model {}", thresholdNodeId, thresholdModelID);
+                ThresholdActionListener thresholdListener = new ThresholdActionListener(
+                    anomalyResultResponse,
+                    featureInResponse,
+                    thresholdModelID,
+                    thresholdNodeId,
+                    detector,
+                    combinedResult,
+                    featureInResponse,
+                    listener,
+                    startTime,
+                    endTime,
+                    adID
+                );
+                transportService
+                    .sendRequest(
+                        thresholdNode.get(),
+                        ThresholdResultAction.NAME,
+                        new ThresholdResultRequest(adID, thresholdModelID, combinedScore),
+                        option,
+                        new ActionListenerResponseHandler<>(thresholdListener, ThresholdResultResponse::new)
+                    );
+            } catch (Exception ex) {
+                handleExecuteException(ex, listener, adID);
+            }
         }
     }
 
@@ -676,35 +708,96 @@ class ThresholdActionListener implements ActionListener<ThresholdResultResponse>
         private List<FeatureData> features;
         private String modelID;
         private AtomicReference<AnomalyDetectionException> failure;
-        private String nodeID;
+        private String thresholdNodeID;
+        private ActionListener<AnomalyResultResponse> listener;
+        private Optional<AnomalyDetector> detector;
+        private CombinedRcfResult combinedResult;
+        private List<FeatureData> featureInResponse;
+        private long startTime;
+        private long endTime;
+        private String adID;
 
         ThresholdActionListener(
             AtomicReference<AnomalyResultResponse> anomalyResultResponse,
             List<FeatureData> features,
             String modelID,
-            AtomicReference<AnomalyDetectionException> failure,
-            String nodeID
+            String thresholdNodeID,
+            Optional<AnomalyDetector> detector,
+            CombinedRcfResult combinedResult,
+            List<FeatureData> featureInResponse,
+            ActionListener<AnomalyResultResponse> listener,
+            long startTime,
+            long endTime,
+            String adID
         ) {
             this.anomalyResultResponse = anomalyResultResponse;
             this.features = features;
             this.modelID = modelID;
-            this.failure = failure;
-            this.nodeID = nodeID;
+            this.thresholdNodeID = thresholdNodeID;
+            this.detector = detector;
+            this.combinedResult = combinedResult;
+            this.featureInResponse = featureInResponse;
+            this.startTime = startTime;
+            this.endTime = endTime;
+            this.failure = new AtomicReference<AnomalyDetectionException>();
+            this.listener = listener;
+            this.adID = adID;
         }
 
         @Override
         public void onResponse(ThresholdResultResponse response) {
             try {
                 anomalyResultResponse.set(new AnomalyResultResponse(response.getAnomalyGrade(), response.getConfidence(), features));
-                stateManager.resetBackpressureCounter(nodeID);
+                stateManager.resetBackpressureCounter(thresholdNodeID);
             } catch (Exception ex) {
-                LOG.error("Unexpected exception", ex);
+                LOG.error("Unexpected exception: {} for {}", ex, adID);
+            } finally {
+                handleThresholdResult();
             }
         }
 
         @Override
         public void onFailure(Exception e) {
-            handlePredictionFailure(e, modelID, nodeID, failure);
+            try {
+                handlePredictionFailure(e, modelID, thresholdNodeID, failure);
+            } catch (Exception ex) {
+                LOG.error("Unexpected exception: {} for {}", ex, adID);
+            } finally {
+                handleThresholdResult();
+            }
+        }
+
+        private void handleThresholdResult() {
+            try {
+                if (coldStartIfNoModel(failure, detector.get())) {
+                    listener.onFailure(new InternalFailure(adID, NO_MODEL_ERR_MSG));
+                    return;
+                }
+
+                if (anomalyResultResponse.get() != null) {
+                    AnomalyResultResponse response = anomalyResultResponse.get();
+                    double confidence = response.getConfidence() * combinedResult.getConfidence();
+                    response = new AnomalyResultResponse(response.getAnomalyGrade(), confidence, response.getFeatures());
+                    listener.onResponse(response);
+                    indexAnomalyResult(
+                        new AnomalyResult(
+                            adID,
+                            Double.valueOf(combinedResult.getScore()),
+                            Double.valueOf(response.getAnomalyGrade()),
+                            Double.valueOf(confidence),
+                            featureInResponse,
+                            Instant.ofEpochMilli(startTime),
+                            Instant.ofEpochMilli(endTime)
+                        )
+                    );
+                } else if (failure.get() != null) {
+                    listener.onFailure(failure.get());
+                } else {
+                    listener.onFailure(new InternalFailure(adID, "Unexpected exception"));
+                }
+            } catch (Exception ex) {
+                handleExecuteException(ex, listener, adID);
+            }
         }
     }
 
diff --git a/src/test/java/com/amazon/opendistroforelasticsearch/ad/AnomalyDetectorJobRunnerTests.java b/src/test/java/com/amazon/opendistroforelasticsearch/ad/AnomalyDetectorJobRunnerTests.java
index f53dff0f..e8aa6c9c 100644
--- a/src/test/java/com/amazon/opendistroforelasticsearch/ad/AnomalyDetectorJobRunnerTests.java
+++ b/src/test/java/com/amazon/opendistroforelasticsearch/ad/AnomalyDetectorJobRunnerTests.java
@@ -116,7 +116,7 @@ public void testRunJobWithNullLockDuration() throws InterruptedException {
     }
 
     @Test
-    public void testRunJobWithLocakDuration() throws InterruptedException {
+    public void testRunJobWithLockDuration() throws InterruptedException {
         when(jobParameter.getLockDurationSeconds()).thenReturn(100L);
         runner.runJob(jobParameter, context);
         Thread.sleep(1000);
diff --git a/src/test/java/com/amazon/opendistroforelasticsearch/ad/transport/AnomalyResultTests.java b/src/test/java/com/amazon/opendistroforelasticsearch/ad/transport/AnomalyResultTests.java
index b7b4ccd0..2e726d9a 100644
--- a/src/test/java/com/amazon/opendistroforelasticsearch/ad/transport/AnomalyResultTests.java
+++ b/src/test/java/com/amazon/opendistroforelasticsearch/ad/transport/AnomalyResultTests.java
@@ -49,7 +49,7 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Function;
 
 import com.amazon.opendistroforelasticsearch.ad.breaker.ADCircuitBreakerService;
@@ -347,7 +347,7 @@ public void testNormal() throws IOException {
         PlainActionFuture<AnomalyResultResponse> listener = new PlainActionFuture<>();
         action.doExecute(null, request, listener);
 
-        AnomalyResultResponse response = listener.actionGet();
+        AnomalyResultResponse response = listener.actionGet(10000L);
         assertAnomalyResultResponse(response, 0, 1, 0d);
     }
 
@@ -755,91 +755,6 @@ public void testRejectRequestBasedOnNegativeCache() {
         assertThat(exception.getMessage(), containsString("There is one query running on AnomalyDetector"));
     }
 
-    public void testRCFLatchAwaitException() throws InterruptedException {
-
-        // These constructors register handler in transport service
-        new RCFResultTransportAction(
-            new ActionFilters(Collections.emptySet()),
-            transportService,
-            normalModelManager,
-            adCircuitBreakerService
-        );
-        new ThresholdResultTransportAction(new ActionFilters(Collections.emptySet()), transportService, normalModelManager);
-
-        AnomalyResultTransportAction action = spy(
-            new AnomalyResultTransportAction(
-                new ActionFilters(Collections.emptySet()),
-                transportService,
-                client,
-                settings,
-                stateManager,
-                runner,
-                anomalyDetectionIndices,
-                featureQuery,
-                normalModelManager,
-                hashRing,
-                clusterService,
-                indexNameResolver,
-                threadPool,
-                adCircuitBreakerService,
-                adStats
-            )
-        );
-
-        CountDownLatch latch = mock(CountDownLatch.class);
-        doThrow(InterruptedException.class).when(latch).await(anyLong(), any(TimeUnit.class));
-        when(action.createCountDownLatch(anyInt())).thenReturn(latch);
-
-        AnomalyResultRequest request = new AnomalyResultRequest(adID, 100, 200);
-        PlainActionFuture<AnomalyResultResponse> listener = new PlainActionFuture<>();
-        action.doExecute(null, request, listener);
-
-        Throwable nestedCause = assertException(listener, AnomalyDetectionException.class);
-        assertEquals(CommonErrorMessages.WAIT_ERR_MSG, nestedCause.getMessage());
-    }
-
-    public void testThresholdLatchAwaitException() throws InterruptedException {
-        // These constructors register handler in transport service
-        new RCFResultTransportAction(
-            new ActionFilters(Collections.emptySet()),
-            transportService,
-            normalModelManager,
-            adCircuitBreakerService
-        );
-        new ThresholdResultTransportAction(new ActionFilters(Collections.emptySet()), transportService, normalModelManager);
-
-        AnomalyResultTransportAction action = spy(
-            new AnomalyResultTransportAction(
-                new ActionFilters(Collections.emptySet()),
-                transportService,
-                client,
-                settings,
-                stateManager,
-                new ColdStartRunner(),
-                anomalyDetectionIndices,
-                featureQuery,
-                normalModelManager,
-                hashRing,
-                clusterService,
-                indexNameResolver,
-                threadPool,
-                adCircuitBreakerService,
-                adStats
-            )
-        );
-
-        CountDownLatch latch = mock(CountDownLatch.class);
-        doThrow(InterruptedException.class).when(latch).await(anyLong(), any(TimeUnit.class));
-        when(action.createCountDownLatch(eq(1))).thenReturn(latch);
-
-        AnomalyResultRequest request = new AnomalyResultRequest(adID, 100, 200);
-        PlainActionFuture<AnomalyResultResponse> listener = new PlainActionFuture<>();
-        action.doExecute(null, request, listener);
-
-        Throwable nestedCause = assertException(listener, AnomalyDetectionException.class);
-        assertEquals(AnomalyResultTransportAction.WAIT_FOR_THRESHOLD_ERR_MSG, nestedCause.getMessage());
-    }
-
     public void alertingRequestTemplate(boolean anomalyResultIndexExists) throws IOException {
         setUpSavingAnomalyResultIndex(anomalyResultIndexExists);
         // These constructors register handler in transport service
@@ -1020,7 +935,9 @@ public void testOnFailureNull() throws IOException {
             adCircuitBreakerService,
             adStats
         );
-        AnomalyResultTransportAction.RCFActionListener listener = action.new RCFActionListener(null, null, null, null);
+        AnomalyResultTransportAction.RCFActionListener listener = action.new RCFActionListener(
+            null, null, null, null, null, null, null, null, null, 0, 0, 0, new AtomicInteger(), null
+        );
         listener.onFailure(null);
     }
 
@@ -1399,7 +1316,9 @@ public void testNullRCFResult() {
             adCircuitBreakerService,
             adStats
         );
-        AnomalyResultTransportAction.RCFActionListener listener = action.new RCFActionListener(null, "123-rcf-0", null, "123");
+        AnomalyResultTransportAction.RCFActionListener listener = action.new RCFActionListener(
+            null, "123-rcf-0", null, "123", null, null, null, null, null, 0, 0, 0, new AtomicInteger(), null
+        );
         listener.onResponse(null);
         assertTrue(testAppender.containsMessage(AnomalyResultTransportAction.NULL_RESPONSE));
     }