Add retention leases replication tests #38857
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,151 @@ | ||
| /* | ||
| * Licensed to Elasticsearch under one or more contributor | ||
| * license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright | ||
| * ownership. Elasticsearch licenses this file to you under | ||
| * the Apache License, Version 2.0 (the "License"); you may | ||
| * not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, | ||
| * software distributed under the License is distributed on an | ||
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY | ||
| * KIND, either express or implied. See the License for the | ||
| * specific language governing permissions and limitations | ||
| * under the License. | ||
| */ | ||
|
|
||
| package org.elasticsearch.index.replication; | ||
|
|
||
| import org.elasticsearch.action.ActionListener; | ||
| import org.elasticsearch.action.support.PlainActionFuture; | ||
| import org.elasticsearch.action.support.replication.ReplicationResponse; | ||
| import org.elasticsearch.cluster.metadata.IndexMetaData; | ||
| import org.elasticsearch.common.Randomness; | ||
| import org.elasticsearch.common.settings.Settings; | ||
| import org.elasticsearch.index.IndexSettings; | ||
| import org.elasticsearch.index.seqno.RetentionLease; | ||
| import org.elasticsearch.index.seqno.RetentionLeaseSyncAction; | ||
| import org.elasticsearch.index.seqno.RetentionLeases; | ||
| import org.elasticsearch.index.shard.IndexShard; | ||
| import org.elasticsearch.index.shard.ShardId; | ||
|
|
||
| import java.util.ArrayList; | ||
| import java.util.List; | ||
| import java.util.concurrent.CountDownLatch; | ||
|
|
||
| import static org.hamcrest.Matchers.containsInAnyOrder; | ||
| import static org.hamcrest.Matchers.equalTo; | ||
| import static org.hamcrest.Matchers.hasSize; | ||
|
|
||
| public class RetentionLeasesReplicationTests extends ESIndexLevelReplicationTestCase { | ||
|
|
||
| public void testSimpleSyncRetentionLeases() throws Exception { | ||
| Settings settings = Settings.builder().put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), true).build(); | ||
| try (ReplicationGroup group = createGroup(between(0, 2), settings)) { | ||
| group.startAll(); | ||
| List<RetentionLease> leases = new ArrayList<>(); | ||
| int iterations = between(1, 100); | ||
| CountDownLatch latch = new CountDownLatch(iterations); | ||
| for (int i = 0; i < iterations; i++) { | ||
| if (leases.isEmpty() == false && rarely()) { | ||
| RetentionLease leaseToRemove = randomFrom(leases); | ||
| leases.remove(leaseToRemove); | ||
| group.removeRetentionLease(leaseToRemove.id(), ActionListener.wrap(latch::countDown)); | ||
| } else { | ||
| RetentionLease newLease = group.addRetentionLease(Integer.toString(i), randomNonNegativeLong(), "test-" + i, | ||
| ActionListener.wrap(latch::countDown)); | ||
| leases.add(newLease); | ||
| } | ||
| } | ||
| RetentionLeases leasesOnPrimary = group.getPrimary().getRetentionLeases(); | ||
| assertThat(leasesOnPrimary.version(), equalTo((long) iterations)); | ||
| assertThat(leasesOnPrimary.primaryTerm(), equalTo(group.getPrimary().getOperationPrimaryTerm())); | ||
| assertThat(leasesOnPrimary.leases(), containsInAnyOrder(leases.toArray(new RetentionLease[0]))); | ||
| latch.await(); | ||
| for (IndexShard replica : group.getReplicas()) { | ||
| assertThat(replica.getRetentionLeases(), equalTo(leasesOnPrimary)); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| public void testOutOfOrderRetentionLeasesRequests() throws Exception { | ||
| Settings settings = Settings.builder().put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), true).build(); | ||
| int numberOfReplicas = between(1, 2); | ||
| IndexMetaData indexMetaData = buildIndexMetaData(numberOfReplicas, settings, indexMapping); | ||
| try (ReplicationGroup group = new ReplicationGroup(indexMetaData) { | ||
| @Override | ||
| protected void syncRetentionLeases(ShardId shardId, RetentionLeases leases, ActionListener<ReplicationResponse> listener) { | ||
| listener.onResponse(new SyncRetentionLeasesResponse(new RetentionLeaseSyncAction.Request(shardId, leases))); | ||
| } | ||
| }) { | ||
| group.startAll(); | ||
| int numLeases = between(1, 10); | ||
| List<RetentionLeaseSyncAction.Request> requests = new ArrayList<>(); | ||
| for (int i = 0; i < numLeases; i++) { | ||
| PlainActionFuture<ReplicationResponse> future = new PlainActionFuture<>(); | ||
| group.addRetentionLease(Integer.toString(i), randomNonNegativeLong(), "test-" + i, future); | ||
| requests.add(((SyncRetentionLeasesResponse) future.actionGet()).syncRequest); | ||
| } | ||
| RetentionLeases leasesOnPrimary = group.getPrimary().getRetentionLeases(); | ||
| for (IndexShard replica : group.getReplicas()) { | ||
| Randomness.shuffle(requests); | ||
| requests.forEach(request -> group.executeRetentionLeasesSyncRequestOnReplica(request, replica)); | ||
| assertThat(replica.getRetentionLeases(), equalTo(leasesOnPrimary)); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| public void testSyncRetentionLeasesWithPrimaryPromotion() throws Exception { | ||
| Settings settings = Settings.builder().put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), true).build(); | ||
| int numberOfReplicas = between(2, 4); | ||
| IndexMetaData indexMetaData = buildIndexMetaData(numberOfReplicas, settings, indexMapping); | ||
| try (ReplicationGroup group = new ReplicationGroup(indexMetaData) { | ||
| @Override | ||
| protected void syncRetentionLeases(ShardId shardId, RetentionLeases leases, ActionListener<ReplicationResponse> listener) { | ||
| listener.onResponse(new SyncRetentionLeasesResponse(new RetentionLeaseSyncAction.Request(shardId, leases))); | ||
| } | ||
| }) { | ||
| group.startAll(); | ||
| int numLeases = between(1, 100); | ||
| IndexShard newPrimary = randomFrom(group.getReplicas()); | ||
| RetentionLeases latestRetentionLeasesOnNewPrimary = RetentionLeases.EMPTY; | ||
| for (int i = 0; i < numLeases; i++) { | ||
| PlainActionFuture<ReplicationResponse> addLeaseFuture = new PlainActionFuture<>(); | ||
| group.addRetentionLease(Integer.toString(i), randomNonNegativeLong(), "test-" + i, addLeaseFuture); | ||
| RetentionLeaseSyncAction.Request request = ((SyncRetentionLeasesResponse) addLeaseFuture.actionGet()).syncRequest; | ||
| for (IndexShard replica : randomSubsetOf(group.getReplicas())) { | ||
| group.executeRetentionLeasesSyncRequestOnReplica(request, replica); | ||
| if (newPrimary == replica) { | ||
| latestRetentionLeasesOnNewPrimary = request.getRetentionLeases(); | ||
| } | ||
| } | ||
| } | ||
| group.promoteReplicaToPrimary(newPrimary).get(); | ||
|
There was a problem hiding this comment. Discuss: should we align the retention-leases when a new primary is promoted? There was a problem hiding this comment. Under what circumstances can the new primary not hold an up-to-date set of leases already? It might perhaps be missing some renewals but I think that's ok. There was a problem hiding this comment. We are adding two new leases to the old primary: L1 and L2. L1 was synced to replica r1; L2 was synced to r2, but the old primary crashed before two leases are properly synced to all two replicas. If any replica is promoted, then the retention leases between copies are not aligned. There was a problem hiding this comment. We sync by copying all the leases from the primary to its replicas, so I don't follow how r2 could receive L2 without also receiving L1 (assuming they were added in this order). However, I think I do see a potential problem:
I think we can prevent this, with peer-recovery retention leases, by insisting that leases do not "go backwards", i.e. they only retain history that is already being retained by another lease. This would mean that C could not discard the history in the situation above because it must already hold a different lease that retains that history. |
||
| // we need to make changes to retention leases to sync it to replicas | ||
| // since we don't sync retention leases when promoting a new primary. | ||
| PlainActionFuture<ReplicationResponse> newLeaseFuture = new PlainActionFuture<>(); | ||
| group.addRetentionLease("new-lease-after-promotion", randomNonNegativeLong(), "test", newLeaseFuture); | ||
| RetentionLeases leasesOnPrimary = group.getPrimary().getRetentionLeases(); | ||
| assertThat(leasesOnPrimary.primaryTerm(), equalTo(group.getPrimary().getOperationPrimaryTerm())); | ||
| assertThat(leasesOnPrimary.version(), equalTo(latestRetentionLeasesOnNewPrimary.version() + 1L)); | ||
| assertThat(leasesOnPrimary.leases(), hasSize(latestRetentionLeasesOnNewPrimary.leases().size() + 1)); | ||
| RetentionLeaseSyncAction.Request request = ((SyncRetentionLeasesResponse) newLeaseFuture.actionGet()).syncRequest; | ||
| for (IndexShard replica : group.getReplicas()) { | ||
| group.executeRetentionLeasesSyncRequestOnReplica(request, replica); | ||
| } | ||
| for (IndexShard replica : group.getReplicas()) { | ||
| assertThat(replica.getRetentionLeases(), equalTo(leasesOnPrimary)); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| static final class SyncRetentionLeasesResponse extends ReplicationResponse { | ||
| final RetentionLeaseSyncAction.Request syncRequest; | ||
| SyncRetentionLeasesResponse(RetentionLeaseSyncAction.Request syncRequest) { | ||
| this.syncRequest = syncRequest; | ||
| } | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
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.
Can we assert that we hold an operation permit here?
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.
The sync action is too thin now, so I inline it directly to the test.