Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Refactor CCS handling code from EsqlSession and IndexResolver into dedicated util class #115976

Closed
wants to merge 5 commits into from
Closed

Refactor CCS handling code from EsqlSession and IndexResolver into dedicated util class #115976

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
c0dcd5f
Initial refactorings. Not the main set yet
quux00 Oct 30, 2024
272eb38
Moved some logic over to EsqlSessionCCSUtils; not complete
quux00 Oct 30, 2024
56ea833
More refactorings.
quux00 Oct 30, 2024
b9d6aef
Removed CcsAwarePreAnalysisActionListener, pushing functionality to E…
quux00 Oct 30, 2024
84163d8
Added tests to EsqlSessionCCSUtils for new methods moved there, such …
quux00 Oct 30, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 5 additions & 8 deletions x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/EsqlExecutionInfo.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -182,6 +182,7 @@ public boolean isSkipUnavailable(String clusterAlias) {
if (RemoteClusterAware.LOCAL_CLUSTER_GROUP_KEY.equals(clusterAlias)) {
return false;
}
// TODO: should first check the skipUn status of each cluster if present in clusterInfo (simplifies testing)
return skipUnavailablePredicate.test(clusterAlias);
}

Expand Down Expand Up @@ -229,7 +230,7 @@ public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params params
b.field(SKIPPED_FIELD.getPreferredName(), getClusterStateCount(Cluster.Status.SKIPPED));
b.field(PARTIAL_FIELD.getPreferredName(), getClusterStateCount(Cluster.Status.PARTIAL));
b.field(FAILED_FIELD.getPreferredName(), getClusterStateCount(Cluster.Status.FAILED));
// each clusterinfo defines its own field object name
// each Cluster object defines its own field object name
b.xContentObject("details", clusterInfo.values().iterator());
});
}
Expand Down Expand Up @@ -352,11 +353,7 @@ public Cluster(
this.successfulShards = successfulShards;
this.skippedShards = skippedShards;
this.failedShards = failedShards;
if (failures == null) {
this.failures = List.of();
} else {
this.failures = failures;
}
this.failures = failures == null ? Collections.emptyList() : failures;
this.took = took;
}

Expand All @@ -373,7 +370,7 @@ public Cluster(StreamInput in) throws IOException {
if (in.getTransportVersion().onOrAfter(TransportVersions.ESQL_CCS_EXEC_INFO_WITH_FAILURES)) {
this.failures = Collections.unmodifiableList(in.readCollectionAsList(ShardSearchFailure::readShardSearchFailure));
} else {
this.failures = List.of();
this.failures = Collections.emptyList();
}
}

Expand Down Expand Up @@ -475,7 +472,7 @@ public Cluster.Builder setTook(TimeValue took) {
@Override
public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
String name = clusterAlias;
if (clusterAlias.equals("")) {
if (clusterAlias.equals(RemoteClusterAware.LOCAL_CLUSTER_GROUP_KEY)) {
name = LOCAL_CLUSTER_NAME_REPRESENTATION;
}
builder.startObject(name);
Expand Down
27 changes: 14 additions & 13 deletions ...k/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/EnrichPolicyResolver.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -289,24 +289,17 @@ public void onResponse(Transport.Connection connection) {
RESOLVE_ACTION_NAME,
new LookupRequest(cluster, remotePolicies),
TransportRequestOptions.EMPTY,
new ActionListenerResponseHandler<>(lookupListener.delegateResponse((l, e) -> {
if (ExceptionsHelper.isRemoteUnavailableException(e)
&& remoteClusterService.isSkipUnavailable(cluster)) {
l.onResponse(new LookupResponse(e));
} else {
l.onFailure(e);
}
}), LookupResponse::new, threadPool.executor(ThreadPool.Names.SEARCH))
new ActionListenerResponseHandler<>(
lookupListener.delegateResponse((l, e) -> failIfSkipUnavailableFalse(e, cluster, l)),
LookupResponse::new,
threadPool.executor(ThreadPool.Names.SEARCH)
)
);
}

@Override
public void onFailure(Exception e) {
if (ExceptionsHelper.isRemoteUnavailableException(e) && remoteClusterService.isSkipUnavailable(cluster)) {
lookupListener.onResponse(new LookupResponse(e));
} else {
lookupListener.onFailure(e);
}
failIfSkipUnavailableFalse(e, cluster, lookupListener);
}
});
}
Expand All @@ -331,6 +324,14 @@ public void onFailure(Exception e) {
}
}

private void failIfSkipUnavailableFalse(Exception e, String cluster, ActionListener<LookupResponse> lookupListener) {
if (ExceptionsHelper.isRemoteUnavailableException(e) && remoteClusterService.isSkipUnavailable(cluster)) {
lookupListener.onResponse(new LookupResponse(e));
} else {
lookupListener.onFailure(e);
}
}

private static class LookupRequest extends TransportRequest {
private final String clusterAlias;
private final Collection<String> policyNames;
Expand Down
Loading