Skip to content

Commit a24a1f5

Browse files
jasontedorjasontedor
authored
Add BWC for retention leases (#39482)
We have to handle the case of a < 6.7.0 and >= 6.7.0 mixed cluster wherein a primary on >= 6.7.0 would otherwise send retention leases to a < 6.7.0 node which would not understand them. This commit adds BWC for this case, and adds a test to ensure that we behave properly here.
1 parent da9f55c commit a24a1f5

File tree

7 files changed

+535
-1
lines changed

7 files changed

+535
-1
lines changed

‎buildSrc/src/main/groovy/org/elasticsearch/gradle/test/ClusterFormationTasks.groovy

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -130,7 +130,9 @@ class ClusterFormationTasks {
130130
if (esConfig.containsKey('discovery.zen.hosts_provider') == false) {
131131
esConfig['discovery.zen.hosts_provider'] = 'file'
132132
}
133-
esConfig['discovery.zen.ping.unicast.hosts'] = []
133+
if (esConfig.containsKey('discovery.zen.ping.unicast.hosts') == false) {
134+
esConfig['discovery.zen.ping.unicast.hosts'] = []
135+
}
134136
esConfig
135137
}
136138
dependsOn = startDependencies

‎qa/retention-lease-bwc/build.gradle

Lines changed: 104 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,104 @@
1+
import org.elasticsearch.gradle.Version
2+
import org.elasticsearch.gradle.test.RestIntegTestTask
3+
4+
/*
5+
* Licensed to Elasticsearch under one or more contributor
6+
* license agreements. See the NOTICE file distributed with
7+
* this work for additional information regarding copyright
8+
* ownership. Elasticsearch licenses this file to you under
9+
* the Apache License, Version 2.0 (the "License"); you may
10+
* not use this file except in compliance with the License.
11+
* You may obtain a copy of the License at
12+
*
13+
* http://www.apache.org/licenses/LICENSE-2.0
14+
*
15+
* Unless required by applicable law or agreed to in writing,
16+
* software distributed under the License is distributed on an
17+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
18+
* KIND, either express or implied. See the License for the
19+
* specific language governing permissions and limitations
20+
* under the License.
21+
*/
22+
23+
apply plugin: 'elasticsearch.esplugin'
24+
25+
esplugin {
26+
description 'add retention lease plugin'
27+
classname 'org.elasticsearch.retention_lease_bwc.AddRetentionLeasePlugin'
28+
}
29+
30+
integTest.enabled = false
31+
32+
task bwcTest {
33+
description = 'runs retention lease backwards compatability tests'
34+
group = 'verification'
35+
}
36+
37+
for (Version version : bwcVersions.wireCompatible) {
38+
if (version.before("6.5.0")) {
39+
// versions before 6.5.0 do not support soft deletes
40+
continue
41+
}
42+
43+
final String baseName = "v${version}"
44+
45+
final Task oldClusterTest = tasks.create(name: "${baseName}#oldClusterTest", type: RestIntegTestTask) {
46+
mustRunAfter(precommit)
47+
includePackaged = false
48+
}
49+
50+
final Object oldClusterTestCluster = extensions.findByName("${baseName}#oldClusterTestCluster")
51+
52+
configure(oldClusterTestCluster) {
53+
numNodes = 2
54+
numBwcNodes = 2
55+
bwcVersion = version
56+
setting "cluster.name", "retention-lease-bwc"
57+
}
58+
59+
final Task newClusterTest = tasks.create(name: "${baseName}#newClusterTest", type: RestIntegTestTask) {
60+
61+
}
62+
63+
final Object newClusterTestCluster = extensions.findByName("${baseName}#newClusterTestCluster")
64+
65+
configure(newClusterTestCluster) {
66+
dependsOn "${baseName}#oldClusterTestCluster#wait"
67+
numNodes = 1
68+
plugin ":qa:retention-lease-bwc"
69+
setting "discovery.zen.ping.unicast.hosts", "\"${-> oldClusterTest.nodes.get(0).transportUri()}\""
70+
setting "cluster.name", "retention-lease-bwc"
71+
setting "node.name", "new-node"
72+
}
73+
74+
final Object newClusterTestRunner = tasks.findByName("${baseName}#newClusterTestRunner")
75+
76+
configure(newClusterTestRunner) {
77+
finalizedBy "${baseName}#oldClusterTestCluster#node0.stop"
78+
finalizedBy "${baseName}#oldClusterTestCluster#node1.stop"
79+
finalizedBy "${baseName}#newClusterTestCluster#stop"
80+
}
81+
82+
final Task versionBwcTest = tasks.create(name: "${baseName}#bwcTest") {
83+
dependsOn newClusterTest
84+
}
85+
86+
if (project.bwc_tests_enabled) {
87+
bwcTest.dependsOn(versionBwcTest)
88+
}
89+
}
90+
91+
task bwcTestSnapshots {
92+
if (project.bwc_tests_enabled) {
93+
for (final def version : bwcVersions.unreleasedWireCompatible) {
94+
// versions before 6.5.0 do not support soft deletes
95+
if (version.before("6.5.0")) {
96+
continue
97+
}
98+
dependsOn "v${version}#bwcTest"
99+
}
100+
}
101+
}
102+
103+
check.dependsOn bwcTestSnapshots
104+
unitTest.enabled = false

‎qa/retention-lease-bwc/src/main/java/org/elasticsearch/retention_lease_bwc/AddRetentionLeasePlugin.java

Lines changed: 51 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,51 @@
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.retention_lease_bwc;
21+
22+
import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
23+
import org.elasticsearch.cluster.node.DiscoveryNodes;
24+
import org.elasticsearch.common.settings.ClusterSettings;
25+
import org.elasticsearch.common.settings.IndexScopedSettings;
26+
import org.elasticsearch.common.settings.Settings;
27+
import org.elasticsearch.common.settings.SettingsFilter;
28+
import org.elasticsearch.plugins.ActionPlugin;
29+
import org.elasticsearch.plugins.Plugin;
30+
import org.elasticsearch.rest.RestController;
31+
import org.elasticsearch.rest.RestHandler;
32+
33+
import java.util.Collections;
34+
import java.util.List;
35+
import java.util.function.Supplier;
36+
37+
public class AddRetentionLeasePlugin extends Plugin implements ActionPlugin {
38+
39+
@Override
40+
public List<RestHandler> getRestHandlers(
41+
final Settings settings,
42+
final RestController restController,
43+
final ClusterSettings clusterSettings,
44+
final IndexScopedSettings indexScopedSettings,
45+
final SettingsFilter settingsFilter,
46+
final IndexNameExpressionResolver indexNameExpressionResolver,
47+
final Supplier<DiscoveryNodes> nodesInCluster) {
48+
return Collections.singletonList(new RestAddRetentionLeaseHandler(settings, restController));
49+
}
50+
51+
}

‎qa/retention-lease-bwc/src/main/java/org/elasticsearch/retention_lease_bwc/RestAddRetentionLeaseHandler.java

Lines changed: 105 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,105 @@
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.retention_lease_bwc;
21+
22+
import org.elasticsearch.action.ActionListener;
23+
import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest;
24+
import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
25+
import org.elasticsearch.action.support.GroupedActionListener;
26+
import org.elasticsearch.client.node.NodeClient;
27+
import org.elasticsearch.cluster.metadata.IndexMetaData;
28+
import org.elasticsearch.common.settings.Settings;
29+
import org.elasticsearch.common.xcontent.XContentBuilder;
30+
import org.elasticsearch.index.seqno.RetentionLeaseActions;
31+
import org.elasticsearch.index.shard.ShardId;
32+
import org.elasticsearch.rest.BaseRestHandler;
33+
import org.elasticsearch.rest.BytesRestResponse;
34+
import org.elasticsearch.rest.RestController;
35+
import org.elasticsearch.rest.RestRequest;
36+
import org.elasticsearch.rest.RestStatus;
37+
import org.elasticsearch.rest.action.RestActionListener;
38+
39+
import java.io.IOException;
40+
import java.io.UncheckedIOException;
41+
import java.util.Collection;
42+
import java.util.Collections;
43+
44+
public class RestAddRetentionLeaseHandler extends BaseRestHandler {
45+
46+
public RestAddRetentionLeaseHandler(final Settings settings, final RestController restController) {
47+
super(settings);
48+
restController.registerHandler(RestRequest.Method.PUT, "/{index}/_add_retention_lease", this);
49+
}
50+
51+
@Override
52+
public String getName() {
53+
return "add_retention_lease";
54+
}
55+
56+
@Override
57+
protected RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
58+
final String index = request.param("index");
59+
final String id = request.param("id");
60+
final long retainingSequenceNumber = Long.parseLong(request.param("retaining_sequence_number"));
61+
final ClusterStateRequest clusterStateRequest = new ClusterStateRequest();
62+
clusterStateRequest.clear();
63+
clusterStateRequest.metaData(true);
64+
clusterStateRequest.indices(index);
65+
return channel ->
66+
client.admin().cluster().state(clusterStateRequest, new ActionListener<ClusterStateResponse>() {
67+
@Override
68+
public void onResponse(final ClusterStateResponse clusterStateResponse) {
69+
final IndexMetaData indexMetaData = clusterStateResponse.getState().metaData().index(index);
70+
final int numberOfShards = IndexMetaData.INDEX_NUMBER_OF_SHARDS_SETTING.get(indexMetaData.getSettings());
71+
72+
final GroupedActionListener<RetentionLeaseActions.Response> listener = new GroupedActionListener<>(
73+
new RestActionListener<Collection<RetentionLeaseActions.Response>>(channel) {
74+
75+
@Override
76+
protected void processResponse(
77+
final Collection<RetentionLeaseActions.Response> responses) throws Exception {
78+
final XContentBuilder builder = channel.newBuilder().startObject().endObject();
79+
channel.sendResponse(new BytesRestResponse(RestStatus.OK, builder));
80+
}
81+
82+
},
83+
numberOfShards,
84+
Collections.emptyList());
85+
for (int i = 0; i < numberOfShards; i++) {
86+
final ShardId shardId = new ShardId(indexMetaData.getIndex(), i);
87+
client.execute(
88+
RetentionLeaseActions.Add.INSTANCE,
89+
new RetentionLeaseActions.AddRequest(shardId, id, retainingSequenceNumber, "rest"),
90+
listener);
91+
}
92+
}
93+
94+
@Override
95+
public void onFailure(final Exception e) {
96+
try {
97+
channel.sendResponse(new BytesRestResponse(channel, RestStatus.SERVICE_UNAVAILABLE, e));
98+
} catch (IOException inner) {
99+
inner.addSuppressed(e);
100+
throw new UncheckedIOException(inner);
101+
}
102+
}
103+
});
104+
}
105+
}

0 commit comments

Comments
 (0)