Fix purging expired replicate requests.
This closes
Newly created async response is added before checking map size nor
purging expired ones. If there are already 100 remaining requests,
the added request will not be executed nor removed.
@@ -316,36 +316,6 @@ public class ThreadPoolRequestReplicator implements RequestReplicator {
throw new IllegalArgumentException("Cannot replicate request to 0 nodes");
}
- // Update headers to indicate the current revision so that we can
- // prevent multiple users changing the flow at the same time
- final Map<String, String> updatedHeaders = new HashMap<>(headers);
- final String requestId = updatedHeaders.computeIfAbsent(REQUEST_TRANSACTION_ID_HEADER, key -> UUID.randomUUID().toString());
-
- // create a response object if one was not already passed to us
- if (response == null) {
- // create the request objects and replicate to all nodes.
- // When the request has completed, we need to ensure that we notify the monitor, if there is one.
- final CompletionCallback completionCallback = clusterResponse -> {
- try {
- onCompletedResponse(requestId);
- } finally {
- if (monitor != null) {
- synchronized (monitor) {
- monitor.notify();
- }
-
- logger.debug("Notified monitor {} because request {} {} has completed", monitor, method, uri);
- }
- }
- };
-
- final Runnable responseConsumedCallback = () -> onResponseConsumed(requestId);
-
- response = new StandardAsyncClusterResponse(requestId, uri, method, nodeIds,
- responseMapper, completionCallback, responseConsumedCallback, merge);
- responseMap.put(requestId, response);
- }
-
// verify all of the nodes exist and are in the proper state
for (final NodeIdentifier nodeId : nodeIds) {
final NodeConnectionStatus status = clusterCoordinator.getConnectionStatus(nodeId);
@@ -360,11 +330,16 @@ public class ThreadPoolRequestReplicator implements RequestReplicator {
logger.debug("Replicating request {} {} with entity {} to {}; response is {}", method, uri, entity, nodeIds, response);
+ // Update headers to indicate the current revision so that we can
+ // prevent multiple users changing the flow at the same time
+ final Map<String, String> updatedHeaders = new HashMap<>(headers);
+ final String requestId = updatedHeaders.computeIfAbsent(REQUEST_TRANSACTION_ID_HEADER, key -> UUID.randomUUID().toString());
+
+ long verifyClusterStateNanos = -1;
if (performVerification) {
final long start = System.nanoTime();
verifyClusterState(method, uri.getPath());
- final long nanos = System.nanoTime() - start;
- response.addTiming("Verify Cluster State", "All Nodes", nanos);
+ verifyClusterStateNanos = System.nanoTime() - start;
}
int numRequests = responseMap.size();
@@ -382,6 +357,35 @@ public class ThreadPoolRequestReplicator implements RequestReplicator {
throw new IllegalStateException("There are too many outstanding HTTP requests with a total " + numRequests + " outstanding requests");
}
+ // create a response object if one was not already passed to us
+ if (response == null) {
+ // create the request objects and replicate to all nodes.
+ // When the request has completed, we need to ensure that we notify the monitor, if there is one.
+ final CompletionCallback completionCallback = clusterResponse -> {
+ try {
+ onCompletedResponse(requestId);
+ } finally {
+ if (monitor != null) {
+ synchronized (monitor) {
+ monitor.notify();
+ }
+
+ logger.debug("Notified monitor {} because request {} {} has completed", monitor, method, uri);
+ }
+ }
+ };
+
+ final Runnable responseConsumedCallback = () -> onResponseConsumed(requestId);
+
+ response = new StandardAsyncClusterResponse(requestId, uri, method, nodeIds,
+ responseMapper, completionCallback, responseConsumedCallback, merge);
+ responseMap.put(requestId, response);
+ }
+
+ if (verifyClusterStateNanos > -1) {
+ response.addTiming("Verify Cluster State", "All Nodes", verifyClusterStateNanos);
+ }
+
logger.debug("For Request ID {}, response object is {}", requestId, response);
// if mutable request, we have to do a two-phase commit where we ask each node to verify
