Add ExecutionInfo to RequestTracker methods #1640
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -16,6 +16,7 @@ | |
| package com.datastax.oss.driver.api.core.tracker; | ||
|
|
||
| import com.datastax.oss.driver.api.core.config.DriverExecutionProfile; | ||
| import com.datastax.oss.driver.api.core.cql.ExecutionInfo; | ||
| import com.datastax.oss.driver.api.core.metadata.Node; | ||
| import com.datastax.oss.driver.api.core.session.Request; | ||
| import com.datastax.oss.driver.api.core.session.Session; | ||
|
|
@@ -35,7 +36,7 @@ public interface RequestTracker extends AutoCloseable { | |
|
|
||
| /** | ||
| * @deprecated This method only exists for backward compatibility. Override {@link | ||
| * #onSuccess(Request, long, DriverExecutionProfile, Node, String, ExecutionInfo)} instead. | ||
| */ | ||
| @Deprecated | ||
| default void onSuccess( | ||
|
|
@@ -44,6 +45,21 @@ default void onSuccess( | |
| @NonNull DriverExecutionProfile executionProfile, | ||
| @NonNull Node node) {} | ||
|
|
||
| /** | ||
| * @deprecated This method only exists for backward compatibility. Override {@link | ||
| * #onSuccess(Request, long, DriverExecutionProfile, Node, String, ExecutionInfo)} instead. | ||
| */ | ||
| @Deprecated | ||
| default void onSuccess( | ||
hhughes marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| @NonNull Request request, | ||
| long latencyNanos, | ||
| @NonNull DriverExecutionProfile executionProfile, | ||
| @NonNull Node node, | ||
| @NonNull String requestLogPrefix) { | ||
| // If client doesn't override onSuccess with requestLogPrefix delegate call to the old method | ||
| onSuccess(request, latencyNanos, executionProfile, node); | ||
| } | ||
|
|
||
| /** | ||
| * Invoked each time a request succeeds. | ||
| * | ||
|
|
@@ -52,20 +68,23 @@ default void onSuccess( | |
| * @param executionProfile the execution profile of this request. | ||
| * @param node the node that returned the successful response. | ||
| * @param requestLogPrefix the dedicated log prefix for this request | ||
| * @param executionInfo the execution info containing the results of this request | ||
| */ | ||
| default void onSuccess( | ||
| @NonNull Request request, | ||
| long latencyNanos, | ||
| @NonNull DriverExecutionProfile executionProfile, | ||
| @NonNull Node node, | ||
| @NonNull String requestLogPrefix, | ||
|
There was a problem hiding this comment. Small request: would it be possible to keep the log prefix as the last parameter? This has been kind of a tacit convention so far. There was a problem hiding this comment. Refactored to make requestLogPrefix the last arg in all the methods. |
||
| @NonNull ExecutionInfo executionInfo) { | ||
| // If client doesn't override onSuccess with executionInfo delegate call to the old method | ||
|
There was a problem hiding this comment. @lukasz-antoniak and I thought we don't need the parameter default void onSuccess(
long latencyNanos,
@NonNull ExecutionInfo executionInfo,
@NonNull String requestLogPrefix) {
onSuccess(executionInfo.getRequest(), latencyNanos, executionInfo.getExecutionProfile(), executionInfo.getCoordinator(), requestLogPrefix);
}If we leave it like it is, the user may be confused if the ones in the parameters and the ones in There was a problem hiding this comment. Good point. As long as we retain and deprecate the old method signatures for backwards-compatability, I'm good. There was a problem hiding this comment. agreed, I like the concise method signature as well! Seeing I think it's fine as proposed though 👍 There was a problem hiding this comment. We may consider putting |
||
| onSuccess(request, latencyNanos, executionProfile, node, requestLogPrefix); | ||
ajweave marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| } | ||
|
|
||
| /** | ||
| * @deprecated This method only exists for backward compatibility. Override {@link | ||
| * #onError(Request, Throwable, long, DriverExecutionProfile, Node, String, ExecutionInfo)} | ||
| * instead. | ||
| */ | ||
| @Deprecated | ||
| default void onError( | ||
|
|
@@ -75,6 +94,23 @@ default void onError( | |
| @NonNull DriverExecutionProfile executionProfile, | ||
| @Nullable Node node) {} | ||
|
|
||
| /** | ||
| * @deprecated This method only exists for backward compatibility. Override {@link | ||
| * #onError(Request, Throwable, long, DriverExecutionProfile, Node, String, ExecutionInfo)} | ||
| * instead. | ||
| */ | ||
| @Deprecated | ||
| default void onError( | ||
hhughes marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| @NonNull Request request, | ||
| @NonNull Throwable error, | ||
| long latencyNanos, | ||
| @NonNull DriverExecutionProfile executionProfile, | ||
| @Nullable Node node, | ||
| @NonNull String requestLogPrefix) { | ||
| // If client doesn't override onError with requestLogPrefix delegate call to the old method | ||
| onError(request, error, latencyNanos, executionProfile, node); | ||
| } | ||
|
|
||
| /** | ||
| * Invoked each time a request fails. | ||
| * | ||
|
|
@@ -83,21 +119,25 @@ default void onError( | |
| * @param executionProfile the execution profile of this request. | ||
| * @param node the node that returned the error response, or {@code null} if the error occurred | ||
| * @param requestLogPrefix the dedicated log prefix for this request | ||
| * @param executionInfo the execution info being returned to the client for this request if | ||
| * available | ||
| */ | ||
| default void onError( | ||
| @NonNull Request request, | ||
| @NonNull Throwable error, | ||
| long latencyNanos, | ||
| @NonNull DriverExecutionProfile executionProfile, | ||
| @Nullable Node node, | ||
| @NonNull String requestLogPrefix, | ||
| @Nullable ExecutionInfo executionInfo) { | ||
| // delegate call to the old method | ||
| onError(request, error, latencyNanos, executionProfile, node, requestLogPrefix); | ||
| } | ||
|
|
||
| /** | ||
| * @deprecated This method only exists for backward compatibility. Override {@link | ||
| * #onNodeError(Request, Throwable, long, DriverExecutionProfile, Node, String, | ||
| * ExecutionInfo)} instead. | ||
| */ | ||
| @Deprecated | ||
| default void onNodeError( | ||
|
|
@@ -107,6 +147,23 @@ default void onNodeError( | |
| @NonNull DriverExecutionProfile executionProfile, | ||
| @NonNull Node node) {} | ||
|
|
||
| /** | ||
| * @deprecated This method only exists for backward compatibility. Override {@link | ||
| * #onNodeError(Request, Throwable, long, DriverExecutionProfile, Node, String, | ||
| * ExecutionInfo)} instead. | ||
| */ | ||
| @Deprecated | ||
| default void onNodeError( | ||
| @NonNull Request request, | ||
| @NonNull Throwable error, | ||
| long latencyNanos, | ||
| @NonNull DriverExecutionProfile executionProfile, | ||
| @NonNull Node node, | ||
| @NonNull String requestLogPrefix) { | ||
| // If client doesn't override onNodeError with requestLogPrefix delegate call to the old method | ||
| onNodeError(request, error, latencyNanos, executionProfile, node); | ||
| } | ||
|
|
||
| /** | ||
| * Invoked each time a request fails at the node level. Similar to {@link #onError(Request, | ||
| * Throwable, long, DriverExecutionProfile, Node, String)} but at a per node level. | ||
|
|
@@ -116,21 +173,24 @@ default void onNodeError( | |
| * @param executionProfile the execution profile of this request. | ||
| * @param node the node that returned the error response. | ||
| * @param requestLogPrefix the dedicated log prefix for this request | ||
| * @param executionInfo the execution info containing the results of this request if available | ||
| */ | ||
| default void onNodeError( | ||
| @NonNull Request request, | ||
| @NonNull Throwable error, | ||
| long latencyNanos, | ||
| @NonNull DriverExecutionProfile executionProfile, | ||
| @NonNull Node node, | ||
| @NonNull String requestLogPrefix, | ||
| @Nullable ExecutionInfo executionInfo) { | ||
| // If client doesn't override onNodeError with requestLogPrefix delegate call to the old method | ||
| onNodeError(request, error, latencyNanos, executionProfile, node, requestLogPrefix); | ||
| } | ||
|
|
||
| /** | ||
| * @deprecated This method only exists for backward compatibility. Override {@link | ||
| * #onNodeSuccess(Request, long, DriverExecutionProfile, Node, String, ExecutionInfo)} | ||
| * instead. | ||
| */ | ||
| @Deprecated | ||
| default void onNodeSuccess( | ||
|
|
@@ -139,6 +199,23 @@ default void onNodeSuccess( | |
| @NonNull DriverExecutionProfile executionProfile, | ||
| @NonNull Node node) {} | ||
|
|
||
| /** | ||
| * @deprecated This method only exists for backward compatibility. Override {@link | ||
| * #onNodeSuccess(Request, long, DriverExecutionProfile, Node, String, ExecutionInfo)} | ||
| * instead. | ||
| */ | ||
| @Deprecated | ||
| default void onNodeSuccess( | ||
hhughes marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| @NonNull Request request, | ||
| long latencyNanos, | ||
| @NonNull DriverExecutionProfile executionProfile, | ||
| @NonNull Node node, | ||
| @NonNull String requestLogPrefix) { | ||
| // If client doesn't override onNodeSuccess with requestLogPrefix delegate call to the old | ||
| // method | ||
| onNodeSuccess(request, latencyNanos, executionProfile, node); | ||
| } | ||
|
|
||
| /** | ||
| * Invoked each time a request succeeds at the node level. Similar to {@link #onSuccess(Request, | ||
| * long, DriverExecutionProfile, Node, String)} but at per node level. | ||
|
|
@@ -148,16 +225,17 @@ default void onNodeSuccess( | |
| * @param executionProfile the execution profile of this request. | ||
| * @param node the node that returned the successful response. | ||
| * @param requestLogPrefix the dedicated log prefix for this request | ||
| * @param executionInfo the execution info containing the results of this request | ||
| */ | ||
| default void onNodeSuccess( | ||
| @NonNull Request request, | ||
| long latencyNanos, | ||
| @NonNull DriverExecutionProfile executionProfile, | ||
| @NonNull Node node, | ||
| @NonNull String requestLogPrefix, | ||
| @NonNull ExecutionInfo executionInfo) { | ||
| // delegate call to the old method | ||
| onNodeSuccess(request, latencyNanos, executionProfile, node, requestLogPrefix); | ||
|
There was a problem hiding this comment. Looks like onSuccess(), onError() and onNodeSuccess() are updated here. We should probably update onNodeError() as well in order to keep the API internally consistent. I don't know that it's actually used in existing driver code beyond MultiplexingRequestTracker but it's still a good idea to keep the API internally consistent. There was a problem hiding this comment. IIRC, from what I could tell, in most error cases no ExecutionInfo is present (but I'm still new to the driver code). I've started going through it again and I think CqlRequestHandler can be refactored to pass it in for some cases. I'll add the new arg without the NonNull annotation. There was a problem hiding this comment. CqlRequestHandler is a bit tangled - there's a NodeResponseCallback inner class with a trackNodeError method and on the containing class, there's a setFinalError method where the ExecutionInfo is created. There was a problem hiding this comment. I agree that CqlRequestHandler does get a little tangled; async code certainly will do that if you don't keep an eye on it :). My comment wasn't connected to the current implementation of CqlRequestHandler, though. Individual request handlers are created by the various request processors. These processors vary widely in implementation, and that's just for the ones we've written; they're defined by an interface so users are always free to write their own! With that in mind I'd say it's important to think about RequestTracker as a general interface which may (or may not) be used by any given handler for any given processor. With that in mind, I'd argue it's important for RequestTracker to present a coherent interface that's consistent across all the on*() handler methods unless there's a very good reason to diverge. There doesn't appear to be any such reason in this case, thus my recommendation to update onNodeError() to keep things consistent. There was a problem hiding this comment. I totally agree on keeping the interface consistent. I've been going through the handlers and for the most part, calls to onNodeError with a null ExecutionInfo are feasible, so that's a good first step. Should a larger refactoring effort to route the ExecutionInfo to those methods be postponed? I suppose a good argument against the refactoring work is that the node-level events in the interface can be considered "in-progress" requests and the final ExecutionInfo instance will be delivered in onSuccess. The only place I see where we can't honor the NonNull contract for onSuccess without significant changes is ContinuousRequestHandlerBase. There was a problem hiding this comment. I figured out a solution in There was a problem hiding this comment. Can we do this in private void trackNodeError(Node node, Throwable error, long nodeResponseTimeNanos) {
if (requestTracker instanceof NoopRequestTracker) {
return;
}
if (nodeResponseTimeNanos == NANOTIME_NOT_MEASURED_YET) {
nodeResponseTimeNanos = System.nanoTime();
}
long latencyNanos = nodeResponseTimeNanos - this.nodeStartTimeNanos;
ExecutionInfo executionInfo =
new DefaultExecutionInfo(
statement,
node,
startedSpeculativeExecutionsCount.get(),
execution,
errors,
null,
null,
true,
session,
context,
executionProfile);
requestTracker.onNodeError(
statement, error, latencyNanos, executionProfile, node, logPrefix, executionInfo);
}There was a problem hiding this comment. I also explored how to pass in ExecutionInfo from ContinuousRequestHandlerBase, see here. How do you think of this approach? There was a problem hiding this comment. Changes to |
||
| } | ||
|
|
||
| /** | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Added an instance variable here for the
ExecutionInfo. It gets populated and later passed to theRequestTrackerincompleteResultSetFuture.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I'm not sure this will always work correctly - the
executionInfofield on theNodeResponseCallbackis populated when the response is processed inprocessResultResponse, but before being used when firing the request tracker methods incompleteResultSetFuture, responses may be queued. If another request comes in before the earlier response is removed from the queue theexecutionInfofield will be overwritten. When the earlier response is later removed from the queue, the execution info provided to the request tracker could be the one from the newer request.Instead of a field attached to the
NodeResponseCallback, perhaps you could use the execution info stored on theresultSetobject inprocessResultResponse? - it's a shame there isn't a common interface (perhaps a useful improvement to file for another time) but it looks like bothContinuousAsyncResultSetandAsyncGraphResultSethave getters for ExecutionInfo objects.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
On further thought, since you only have an
Objectat this stage, I don't know how you would get an execution info for the request tracker call if you end up with a something that isn't aContinuousAsyncResultSetorAsyncGraphResultSetso perhaps the solution I suggested above isn't going to work.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I apologize for calling out details you no doubt already know off the top of your head, it's just for my own benefit being new to the code.
This is a bit of a pickle.
ResultSetTis just anObjectand it looks like just sharpening up the definition of that type parameter is impossible because there's not a unified type structure for the different kinds of results in the internal API.I could add a type check and cast for both
AsyncPagingIterableandAsyncGraphResultSetwhich both provide agetExecutionInfomethod.If that works, then we might want to go a bit further and refactor a bit to add a marking interface or something to prevent other types from being passed in, but that seems like an extensive change.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I removed the
ExecutionInfoinstance variable and added type checks and casts for bothAsyncPagingIterableandAsyncGraphResultSetto get theExecutionInfofrom there.I attempted to run the
ContinuousPagingITtest with DSE to validate, but I'm having trouble with CCM. The CCM node fails to start with this error:I'm setting these Java options:
Any tips for running the integration test locally are appreciated.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
It's a bit of work to set up. I ran all the tests against this branch, and it passed all the tests.