Skip to content

Commit 21d52f0

Browse files
authored
Ensure that local cluster alias is never treated as remote (#37121)
With #36997 we added support for providing a local cluster alias with a `SearchRequest`. We intended to make sure that when provided as part of a search request, the cluster alias would never be used for connection lookups. Yet due to a bug we would still end up looking up the connection from the remote ones. This commit adds a test to make sure that whenever we set the cluster alias to the `SearchRequest` (which can only be done at transport), such alias is used as index prefix in the returned hits. No errors are thrown despite no remote clusters are configured indicating that such alias is never used for connection look-ups. Also, we add explicit support for the empty cluster alias when printing out index names through `RemoteClusterAware#buildRemoteIndexName`. In fact we don't want to print out `:index` when the cluster alias is set to empty string, but rather `index`. Yet, the semantic of empty string is different compared to `null` as it will still disable final reduction. This will be used in CCS when searching against remote clusters as well as the local one, the local one will have empty prefix yet it will need to disable final reduction so that its results will be properly merged with the ones coming from the remote clusters.
1 parent 0fd27d4 commit 21d52f0

File tree

5 files changed

+111
-5
lines changed

5 files changed

+111
-5
lines changed

server/src/main/java/org/elasticsearch/action/search/SearchTransportService.java

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
import org.elasticsearch.action.support.HandledTransportAction.ChannelActionListener;
2727
import org.elasticsearch.action.support.IndicesOptions;
2828
import org.elasticsearch.cluster.node.DiscoveryNode;
29+
import org.elasticsearch.common.Nullable;
2930
import org.elasticsearch.common.io.stream.StreamInput;
3031
import org.elasticsearch.common.io.stream.StreamOutput;
3132
import org.elasticsearch.common.io.stream.Writeable;
@@ -401,11 +402,11 @@ public void onFailure(Exception e) {
401402
/**
402403
* Returns a connection to the given node on the provided cluster. If the cluster alias is <code>null</code> the node will be resolved
403404
* against the local cluster.
404-
* @param clusterAlias the cluster alias the node should be resolve against
405+
* @param clusterAlias the cluster alias the node should be resolved against
405406
* @param node the node to resolve
406407
* @return a connection to the given node belonging to the cluster with the provided alias.
407408
*/
408-
Transport.Connection getConnection(String clusterAlias, DiscoveryNode node) {
409+
Transport.Connection getConnection(@Nullable String clusterAlias, DiscoveryNode node) {
409410
if (clusterAlias == null) {
410411
return transportService.getConnection(node);
411412
} else {

server/src/main/java/org/elasticsearch/action/search/TransportSearchAction.java

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -353,16 +353,19 @@ static BiFunction<String, String, Transport.Connection> buildConnectionLookup(St
353353
BiFunction<String, DiscoveryNode, Transport.Connection> nodeToConnection) {
354354
return (clusterAlias, nodeId) -> {
355355
final DiscoveryNode discoveryNode;
356+
final boolean remoteCluster;
356357
if (clusterAlias == null || requestClusterAlias != null) {
357358
assert requestClusterAlias == null || requestClusterAlias.equals(clusterAlias);
358359
discoveryNode = localNodes.apply(nodeId);
360+
remoteCluster = false;
359361
} else {
360362
discoveryNode = remoteNodes.apply(clusterAlias, nodeId);
363+
remoteCluster = true;
361364
}
362365
if (discoveryNode == null) {
363366
throw new IllegalStateException("no node found for id: " + nodeId);
364367
}
365-
return nodeToConnection.apply(clusterAlias, discoveryNode);
368+
return nodeToConnection.apply(remoteCluster ? clusterAlias : null, discoveryNode);
366369
};
367370
}
368371

server/src/main/java/org/elasticsearch/transport/RemoteClusterAware.java

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -346,7 +346,7 @@ private static int indexOfPortSeparator(String remoteHost) {
346346
}
347347

348348
public static String buildRemoteIndexName(String clusterAlias, String indexName) {
349-
return clusterAlias != null ? clusterAlias + REMOTE_CLUSTER_INDEX_SEPARATOR + indexName : indexName;
349+
return clusterAlias == null || LOCAL_CLUSTER_GROUP_KEY.equals(clusterAlias)
350+
? indexName : clusterAlias + REMOTE_CLUSTER_INDEX_SEPARATOR + indexName;
350351
}
351-
352352
}
Lines changed: 62 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,62 @@
1+
/*
2+
* Licensed to Elasticsearch under one or more contributor
3+
* license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright
5+
* ownership. Elasticsearch licenses this file to you under
6+
* the Apache License, Version 2.0 (the "License"); you may
7+
* not use this file except in compliance with the License.
8+
* You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package org.elasticsearch.action.search;
21+
22+
import org.elasticsearch.action.index.IndexRequest;
23+
import org.elasticsearch.action.index.IndexResponse;
24+
import org.elasticsearch.action.support.WriteRequest;
25+
import org.elasticsearch.rest.RestStatus;
26+
import org.elasticsearch.search.SearchHit;
27+
import org.elasticsearch.test.ESSingleNodeTestCase;
28+
29+
public class TransportSearchActionSingleNodeTests extends ESSingleNodeTestCase {
30+
31+
public void testLocalClusterAlias() {
32+
IndexRequest indexRequest = new IndexRequest("test");
33+
indexRequest.id("1");
34+
indexRequest.source("field", "value");
35+
indexRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL);
36+
IndexResponse indexResponse = client().index(indexRequest).actionGet();
37+
assertEquals(RestStatus.CREATED, indexResponse.status());
38+
39+
{
40+
SearchRequest searchRequest = new SearchRequest("local");
41+
SearchResponse searchResponse = client().search(searchRequest).actionGet();
42+
assertEquals(1, searchResponse.getHits().getTotalHits().value);
43+
SearchHit[] hits = searchResponse.getHits().getHits();
44+
assertEquals(1, hits.length);
45+
SearchHit hit = hits[0];
46+
assertEquals("local", hit.getClusterAlias());
47+
assertEquals("test", hit.getIndex());
48+
assertEquals("1", hit.getId());
49+
}
50+
{
51+
SearchRequest searchRequest = new SearchRequest("");
52+
SearchResponse searchResponse = client().search(searchRequest).actionGet();
53+
assertEquals(1, searchResponse.getHits().getTotalHits().value);
54+
SearchHit[] hits = searchResponse.getHits().getHits();
55+
assertEquals(1, hits.length);
56+
SearchHit hit = hits[0];
57+
assertEquals("", hit.getClusterAlias());
58+
assertEquals("test", hit.getIndex());
59+
assertEquals("1", hit.getId());
60+
}
61+
}
62+
}
Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,40 @@
1+
/*
2+
* Licensed to Elasticsearch under one or more contributor
3+
* license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright
5+
* ownership. Elasticsearch licenses this file to you under
6+
* the Apache License, Version 2.0 (the "License"); you may
7+
* not use this file except in compliance with the License.
8+
* You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package org.elasticsearch.transport;
21+
22+
import org.elasticsearch.test.ESTestCase;
23+
24+
public class RemoteClusterAwareTests extends ESTestCase {
25+
26+
public void testBuildRemoteIndexName() {
27+
{
28+
String clusterAlias = randomAlphaOfLengthBetween(5, 10);
29+
String index = randomAlphaOfLengthBetween(5, 10);
30+
String remoteIndexName = RemoteClusterAware.buildRemoteIndexName(clusterAlias, index);
31+
assertEquals(clusterAlias + ":" + index, remoteIndexName);
32+
}
33+
{
34+
String index = randomAlphaOfLengthBetween(5, 10);
35+
String clusterAlias = randomBoolean() ? null : RemoteClusterAware.LOCAL_CLUSTER_GROUP_KEY;
36+
String remoteIndexName = RemoteClusterAware.buildRemoteIndexName(clusterAlias, index);
37+
assertEquals(index, remoteIndexName);
38+
}
39+
}
40+
}

0 commit comments

Comments
 (0)