Skip to content

Commit

Permalink
Add ExecutionInfo to RequestTracker callbacks
Browse files Browse the repository at this point in the history
  • Loading branch information
lukasz-antoniak committed Aug 30, 2024
1 parent 43db6f6 commit 51fceda
Show file tree
Hide file tree
Showing 10 changed files with 423 additions and 255 deletions.
10 changes: 10 additions & 0 deletions core/revapi.json
Original file line number Diff line number Diff line change
Expand Up @@ -6956,6 +6956,16 @@
"old": "method java.lang.Throwable java.lang.Throwable::fillInStackTrace() @ com.fasterxml.jackson.databind.deser.UnresolvedForwardReference",
"new": "method com.fasterxml.jackson.databind.deser.UnresolvedForwardReference com.fasterxml.jackson.databind.deser.UnresolvedForwardReference::fillInStackTrace()",
"justification": "Upgrade jackson-databind to 2.13.4.1 to address CVEs, API change cause: https://github.com/FasterXML/jackson-databind/issues/3419"
},
{
"code": "java.method.addedToInterface",
"new": "method com.datastax.oss.driver.api.core.config.DriverExecutionProfile com.datastax.dse.driver.api.core.graph.GraphExecutionInfo::getExecutionProfile()",
"justification": "Execution profile is frequently passed together with ExecutionInfo, so to improve API clarity it can be included inside the object. Part of refactoring to add ExecutionInfo object to RequestTracker callbacks."
},
{
"code": "java.method.addedToInterface",
"new": "method com.datastax.oss.driver.api.core.config.DriverExecutionProfile com.datastax.oss.driver.api.core.cql.ExecutionInfo::getExecutionProfile()",
"justification": "Execution profile is frequently passed together with ExecutionInfo, so to improve API clarity it can be included inside the object. Part of refactoring to add ExecutionInfo object to RequestTracker callbacks."
}
]
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
package com.datastax.dse.driver.api.core.graph;

import com.datastax.oss.driver.api.core.DefaultProtocolVersion;
import com.datastax.oss.driver.api.core.config.DriverExecutionProfile;
import com.datastax.oss.driver.api.core.metadata.Node;
import com.datastax.oss.driver.api.core.specex.SpeculativeExecutionPolicy;
import java.nio.ByteBuffer;
Expand All @@ -37,6 +38,9 @@ public interface GraphExecutionInfo {
/** The statement that was executed. */
GraphStatement<?> getStatement();

/** @return Execution profile applied when executing given request. */
DriverExecutionProfile getExecutionProfile();

/** The node that was used as a coordinator to successfully complete the query. */
Node getCoordinator();

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -833,7 +833,8 @@ public void onFailure(@NonNull Throwable error) {
private void processResultResponse(@NonNull Result result, @Nullable Frame frame) {
assert lock.isHeldByCurrentThread();
try {
ExecutionInfo executionInfo = createExecutionInfo(result, frame);
ExecutionInfo executionInfo =
createExecutionInfo().withServerResponse(result, frame).build();
if (result instanceof Rows) {
DseRowsMetadata rowsMetadata = (DseRowsMetadata) ((Rows) result).getMetadata();
if (columnDefinitions == null) {
Expand Down Expand Up @@ -1458,7 +1459,7 @@ private void trackNodeError(
latencyNanos,
executionProfile,
node,
createExecutionInfo(frame),
createExecutionInfo().withServerResponse(frame).build(),
logPrefix);
}
}
Expand Down Expand Up @@ -1576,7 +1577,7 @@ private void completeResultSetFuture(

ExecutionInfo executionInfo = null;
if (pageOrError instanceof AsyncPagingIterable) {
executionInfo = ((AsyncPagingIterable) pageOrError).getExecutionInfo();
executionInfo = ((AsyncPagingIterable<?, ?>) pageOrError).getExecutionInfo();
} else if (pageOrError instanceof AsyncGraphResultSet) {
executionInfo = ((AsyncGraphResultSet) pageOrError).getRequestExecutionInfo();
}
Expand Down Expand Up @@ -1612,34 +1613,13 @@ private void completeResultSetFuture(
}

@NonNull
private ExecutionInfo createExecutionInfo(@NonNull Result result, @Nullable Frame response) {
ByteBuffer pagingState =
result instanceof Rows ? ((Rows) result).getMetadata().pagingState : null;
return new DefaultExecutionInfo(
private DefaultExecutionInfo.Builder createExecutionInfo() {
return DefaultExecutionInfo.builder(
statement,
node,
startedSpeculativeExecutionsCount.get(),
executionIndex,
errors,
pagingState,
response,
true,
session,
context,
executionProfile);
}

@NonNull
private ExecutionInfo createExecutionInfo(@Nullable Frame response) {
return new DefaultExecutionInfo(
statement,
node,
startedSpeculativeExecutionsCount.get(),
executionIndex,
errors,
null,
response,
true,
session,
context,
executionProfile);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
package com.datastax.dse.driver.internal.core.graph;

import com.datastax.dse.driver.api.core.graph.GraphStatement;
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.cql.QueryTrace;
import com.datastax.oss.driver.api.core.cql.Statement;
Expand Down Expand Up @@ -62,6 +63,11 @@ public Statement<?> getStatement() {
throw new ClassCastException("GraphStatement cannot be cast to Statement");
}

@Override
public DriverExecutionProfile getExecutionProfile() {
return graphExecutionInfo.getExecutionProfile();
}

@Nullable
@Override
public Node getCoordinator() {
Expand Down Expand Up @@ -146,6 +152,11 @@ public GraphStatement<?> getStatement() {
return (GraphStatement<?>) executionInfo.getRequest();
}

@Override
public DriverExecutionProfile getExecutionProfile() {
return executionInfo.getExecutionProfile();
}

@Override
public Node getCoordinator() {
return executionInfo.getCoordinator();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -329,7 +329,18 @@ private void cancelScheduledTasks() {
private void setFinalResult(
Result resultMessage, Frame responseFrame, NodeResponseCallback callback) {
try {
ExecutionInfo executionInfo = buildExecutionInfo(callback, responseFrame);
ExecutionInfo executionInfo =
DefaultExecutionInfo.builder(
callback.statement,
callback.node,
startedSpeculativeExecutionsCount.get(),
callback.execution,
errors,
session,
context,
callback.executionProfile)
.withServerResponse(resultMessage, responseFrame)
.build();
DriverExecutionProfile executionProfile =
Conversions.resolveExecutionProfile(callback.statement, context);
GraphProtocol subProtocol =
Expand Down Expand Up @@ -426,23 +437,6 @@ private void logServerWarnings(GraphStatement<?> statement, List<String> warning
LOG.warn("Query '{}' generated server side warning(s): {}", statementString, warning));
}

private ExecutionInfo buildExecutionInfo(NodeResponseCallback callback, Frame responseFrame) {
DriverExecutionProfile executionProfile =
Conversions.resolveExecutionProfile(callback.statement, context);
return new DefaultExecutionInfo(
callback.statement,
callback.node,
startedSpeculativeExecutionsCount.get(),
callback.execution,
errors,
null,
responseFrame,
true,
session,
context,
executionProfile);
}

@Override
public void onThrottleFailure(@NonNull RequestThrottlingException error) {
DriverExecutionProfile executionProfile =
Expand All @@ -457,18 +451,16 @@ private void setFinalError(
DriverExecutionProfile executionProfile =
Conversions.resolveExecutionProfile(statement, context);
ExecutionInfo executionInfo =
new DefaultExecutionInfo(
statement,
node,
startedSpeculativeExecutionsCount.get(),
execution,
errors,
null,
null,
true,
session,
context,
executionProfile);
DefaultExecutionInfo.builder(
statement,
node,
startedSpeculativeExecutionsCount.get(),
execution,
errors,
session,
context,
executionProfile)
.build();
if (error instanceof DriverException) {
((DriverException) error).setExecutionInfo(executionInfo);
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
import com.datastax.oss.driver.api.core.DriverException;
import com.datastax.oss.driver.api.core.RequestThrottlingException;
import com.datastax.oss.driver.api.core.config.DefaultDriverOption;
import com.datastax.oss.driver.api.core.config.DriverExecutionProfile;
import com.datastax.oss.driver.api.core.detach.AttachmentPoint;
import com.datastax.oss.driver.api.core.metadata.Node;
import com.datastax.oss.driver.api.core.retry.RetryDecision;
Expand Down Expand Up @@ -66,6 +67,9 @@ default Request getRequest() {
@Deprecated
Statement<?> getStatement();

/** @return Execution profile applied when executing given request. */
DriverExecutionProfile getExecutionProfile();

/**
* The node that acted as a coordinator for the query.
*
Expand Down
Loading

0 comments on commit 51fceda

Please sign in to comment.