honeymoose
/
OpenSearch
mirror of https://github.com/honeymoose/OpenSearch.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Use peer recovery retention leases for indices without soft-deletes (#50351) 

Today, the replica allocator uses peer recovery retention leases to
select the best-matched copies when allocating replicas of indices with
soft-deletes. We can employ this mechanism for indices without
soft-deletes because the retaining sequence number of a PRRL is the
persisted global checkpoint (plus one) of that copy. If the primary and
replica have the same retaining sequence number, then we should be able
to perform a noop recovery. The reason is that we must be retaining
translog up to the local checkpoint of the safe commit, which is at most
the global checkpoint of either copy). The only limitation is that we
might not cancel ongoing file-based recoveries with PRRLs for noop
recoveries. We can't make the translog retention policy comply with
PRRLs. We also have this problem with soft-deletes if a PRRL is about to
expire.

Relates #45136
Relates #46959
This commit is contained in:
Nhat Nguyen 2019-12-20 00:39:53 -05:00
parent 1dc98ad617
commit 33204c2055
11 changed files with 89 additions and 119 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
qa/full-cluster-restart/src/test/java/org/elasticsearch/upgrades/FullClusterRestartIT.java
View File

@ -1404,6 +1404,7 @@ public class FullClusterRestartIT extends AbstractFullClusterRestartTestCase {
} else { } else {
ensureGreen(index); ensureGreen(index);
assertNoFileBasedRecovery(index, n -> true); assertNoFileBasedRecovery(index, n -> true);
ensurePeerRecoveryRetentionLeasesRenewedAndSynced(index);
} }
} }
@ -1429,6 +1430,7 @@ public class FullClusterRestartIT extends AbstractFullClusterRestartTestCase {
ensureGreen(index); ensureGreen(index);
flush(index, true); flush(index, true);
assertEmptyTranslog(index); assertEmptyTranslog(index);
ensurePeerRecoveryRetentionLeasesRenewedAndSynced(index);
} }
} }
} }

22
qa/rolling-upgrade/src/test/java/org/elasticsearch/upgrades/RecoveryIT.java
View File

@ -540,28 +540,6 @@ public class RecoveryIT extends AbstractRollingTestCase {
return Version.fromId(Integer.parseInt(ObjectPath.createFromResponse(response).evaluate(versionCreatedSetting))); return Version.fromId(Integer.parseInt(ObjectPath.createFromResponse(response).evaluate(versionCreatedSetting)));
} }
/**
* Returns the minimum node version among all nodes of the cluster
*/
private static Version minimumNodeVersion() throws IOException {
final Request request = new Request("GET", "_nodes");
request.addParameter("filter_path", "nodes.*.version");
final Response response = client().performRequest(request);
final Map<String, Object> nodes = ObjectPath.createFromResponse(response).evaluate("nodes");
Version minVersion = null;
for (Map.Entry<String, Object> node : nodes.entrySet()) {
@SuppressWarnings("unchecked")
Version nodeVersion = Version.fromString((String) ((Map<String, Object>) node.getValue()).get("version"));
if (minVersion == null || minVersion.after(nodeVersion)) {
minVersion = nodeVersion;
}
}
assertNotNull(minVersion);
return minVersion;
}
/** /**
* Asserts that an index is closed in the cluster state. If `checkRoutingTable` is true, it also asserts * Asserts that an index is closed in the cluster state. If `checkRoutingTable` is true, it also asserts
* that the index has started shards. * that the index has started shards.

2
server/src/main/java/org/elasticsearch/index/IndexService.java
View File

@ -824,10 +824,8 @@ public class IndexService extends AbstractIndexComponent implements IndicesClust
} }
private void syncRetentionLeases() { private void syncRetentionLeases() {
if (indexSettings.isSoftDeleteEnabled()) {
sync(IndexShard::syncRetentionLeases, "retention lease"); sync(IndexShard::syncRetentionLeases, "retention lease");
} }
}
private void sync(final Consumer<IndexShard> sync, final String source) { private void sync(final Consumer<IndexShard> sync, final String source) {
for (final IndexShard shard : this.shards.values()) { for (final IndexShard shard : this.shards.values()) {

13
server/src/main/java/org/elasticsearch/index/seqno/ReplicationTracker.java
View File

@ -906,10 +906,10 @@ public class ReplicationTracker extends AbstractIndexShardComponent implements L
this.pendingInSync = new HashSet<>(); this.pendingInSync = new HashSet<>();
this.routingTable = null; this.routingTable = null;
this.replicationGroup = null; this.replicationGroup = null;
this.hasAllPeerRecoveryRetentionLeases = indexSettings.isSoftDeleteEnabled() && this.hasAllPeerRecoveryRetentionLeases = indexSettings.getIndexVersionCreated().onOrAfter(Version.V_7_6_0) ||
(indexSettings.getIndexVersionCreated().onOrAfter(Version.V_7_6_0) || (indexSettings.isSoftDeleteEnabled() &&
(indexSettings.getIndexVersionCreated().onOrAfter(Version.V_7_4_0) && indexSettings.getIndexVersionCreated().onOrAfter(Version.V_7_4_0) &&
indexSettings.getIndexMetaData().getState() == IndexMetaData.State.OPEN)); indexSettings.getIndexMetaData().getState() == IndexMetaData.State.OPEN);
this.fileBasedRecoveryThreshold = IndexSettings.FILE_BASED_RECOVERY_THRESHOLD_SETTING.get(indexSettings.getSettings()); this.fileBasedRecoveryThreshold = IndexSettings.FILE_BASED_RECOVERY_THRESHOLD_SETTING.get(indexSettings.getSettings());
this.safeCommitInfoSupplier = safeCommitInfoSupplier; this.safeCommitInfoSupplier = safeCommitInfoSupplier;
assert Version.V_EMPTY.equals(indexSettings.getIndexVersionCreated()) == false; assert Version.V_EMPTY.equals(indexSettings.getIndexVersionCreated()) == false;
@ -1005,10 +1005,7 @@ public class ReplicationTracker extends AbstractIndexShardComponent implements L
updateLocalCheckpoint(shardAllocationId, checkpoints.get(shardAllocationId), localCheckpoint); updateLocalCheckpoint(shardAllocationId, checkpoints.get(shardAllocationId), localCheckpoint);
updateGlobalCheckpointOnPrimary(); updateGlobalCheckpointOnPrimary();
if (indexSettings.isSoftDeleteEnabled()) {
addPeerRecoveryRetentionLeaseForSolePrimary(); addPeerRecoveryRetentionLeaseForSolePrimary();
}
assert invariant(); assert invariant();
} }
@ -1373,7 +1370,7 @@ public class ReplicationTracker extends AbstractIndexShardComponent implements L
* prior to {@link Version#V_7_4_0} that does not create peer-recovery retention leases. * prior to {@link Version#V_7_4_0} that does not create peer-recovery retention leases.
*/ */
public synchronized void createMissingPeerRecoveryRetentionLeases(ActionListener<Void> listener) { public synchronized void createMissingPeerRecoveryRetentionLeases(ActionListener<Void> listener) {
if (indexSettings().isSoftDeleteEnabled() && hasAllPeerRecoveryRetentionLeases == false) { if (hasAllPeerRecoveryRetentionLeases == false) {
final List<ShardRouting> shardRoutings = routingTable.assignedShards(); final List<ShardRouting> shardRoutings = routingTable.assignedShards();
final GroupedActionListener<ReplicationResponse> groupedActionListener = new GroupedActionListener<>(ActionListener.wrap(vs -> { final GroupedActionListener<ReplicationResponse> groupedActionListener = new GroupedActionListener<>(ActionListener.wrap(vs -> {
setHasAllPeerRecoveryRetentionLeases(); setHasAllPeerRecoveryRetentionLeases();

9
server/src/main/java/org/elasticsearch/index/shard/IndexShard.java
View File

@ -1917,10 +1917,10 @@ public class IndexShard extends AbstractIndexShardComponent implements IndicesCl
public void onSettingsChanged() { public void onSettingsChanged() {
Engine engineOrNull = getEngineOrNull(); Engine engineOrNull = getEngineOrNull();
if (engineOrNull != null) { if (engineOrNull != null) {
final boolean useRetentionLeasesInPeerRecovery = this.useRetentionLeasesInPeerRecovery; final boolean disableTranslogRetention = indexSettings.isSoftDeleteEnabled() && useRetentionLeasesInPeerRecovery;
engineOrNull.onSettingsChanged( engineOrNull.onSettingsChanged(
useRetentionLeasesInPeerRecovery ? TimeValue.MINUS_ONE : indexSettings.getTranslogRetentionAge(), disableTranslogRetention ? TimeValue.MINUS_ONE : indexSettings.getTranslogRetentionAge(),
useRetentionLeasesInPeerRecovery ? new ByteSizeValue(-1) : indexSettings.getTranslogRetentionSize(), disableTranslogRetention ? new ByteSizeValue(-1) : indexSettings.getTranslogRetentionSize(),
indexSettings.getSoftDeleteRetentionOperations() indexSettings.getSoftDeleteRetentionOperations()
); );
} }
@ -2249,7 +2249,6 @@ public class IndexShard extends AbstractIndexShardComponent implements IndicesCl
public void syncRetentionLeases() { public void syncRetentionLeases() {
assert assertPrimaryMode(); assert assertPrimaryMode();
verifyNotClosed(); verifyNotClosed();
ensureSoftDeletesEnabled("retention leases");
replicationTracker.renewPeerRecoveryRetentionLeases(); replicationTracker.renewPeerRecoveryRetentionLeases();
final Tuple<Boolean, RetentionLeases> retentionLeases = getRetentionLeases(true); final Tuple<Boolean, RetentionLeases> retentionLeases = getRetentionLeases(true);
if (retentionLeases.v1()) { if (retentionLeases.v1()) {
@ -2646,7 +2645,7 @@ public class IndexShard extends AbstractIndexShardComponent implements IndicesCl
ActionListener<ReplicationResponse> listener) { ActionListener<ReplicationResponse> listener) {
assert assertPrimaryMode(); assert assertPrimaryMode();
// only needed for BWC reasons involving rolling upgrades from versions that do not support PRRLs: // only needed for BWC reasons involving rolling upgrades from versions that do not support PRRLs:
assert indexSettings.getIndexVersionCreated().before(Version.V_7_4_0); assert indexSettings.getIndexVersionCreated().before(Version.V_7_4_0) || indexSettings.isSoftDeleteEnabled() == false;
return replicationTracker.addPeerRecoveryRetentionLease(nodeId, globalCheckpoint, listener); return replicationTracker.addPeerRecoveryRetentionLease(nodeId, globalCheckpoint, listener);
} }

36
server/src/main/java/org/elasticsearch/indices/recovery/RecoverySourceHandler.java
View File

@ -167,12 +167,12 @@ public class RecoverySourceHandler {
throw new DelayRecoveryException("source node does not have the shard listed in its state as allocated on the node"); throw new DelayRecoveryException("source node does not have the shard listed in its state as allocated on the node");
} }
assert targetShardRouting.initializing() : "expected recovery target to be initializing but was " + targetShardRouting; assert targetShardRouting.initializing() : "expected recovery target to be initializing but was " + targetShardRouting;
retentionLeaseRef.set(softDeletesEnabled ? shard.getRetentionLeases().get( retentionLeaseRef.set(
ReplicationTracker.getPeerRecoveryRetentionLeaseId(targetShardRouting)) : null); shard.getRetentionLeases().get(ReplicationTracker.getPeerRecoveryRetentionLeaseId(targetShardRouting)));
}, shardId + " validating recovery target ["+ request.targetAllocationId() + "] registered ", }, shardId + " validating recovery target ["+ request.targetAllocationId() + "] registered ",
shard, cancellableThreads, logger); shard, cancellableThreads, logger);
final Engine.HistorySource historySource; final Engine.HistorySource historySource;
if (shard.useRetentionLeasesInPeerRecovery() || retentionLeaseRef.get() != null) { if (softDeletesEnabled && (shard.useRetentionLeasesInPeerRecovery() || retentionLeaseRef.get() != null)) {
historySource = Engine.HistorySource.INDEX; historySource = Engine.HistorySource.INDEX;
} else { } else {
historySource = Engine.HistorySource.TRANSLOG; historySource = Engine.HistorySource.TRANSLOG;
@ -192,7 +192,7 @@ public class RecoverySourceHandler {
// Also it's pretty cheap when soft deletes are enabled, and it'd be a disaster if we tried a sequence-number-based recovery // Also it's pretty cheap when soft deletes are enabled, and it'd be a disaster if we tried a sequence-number-based recovery
// without having a complete history. // without having a complete history.
if (isSequenceNumberBasedRecovery && retentionLeaseRef.get() != null) { if (isSequenceNumberBasedRecovery && softDeletesEnabled && retentionLeaseRef.get() != null) {
// all the history we need is retained by an existing retention lease, so we do not need a separate retention lock // all the history we need is retained by an existing retention lease, so we do not need a separate retention lock
retentionLock.close(); retentionLock.close();
logger.trace("history is retained by {}", retentionLeaseRef.get()); logger.trace("history is retained by {}", retentionLeaseRef.get());
@ -211,7 +211,7 @@ public class RecoverySourceHandler {
if (isSequenceNumberBasedRecovery) { if (isSequenceNumberBasedRecovery) {
logger.trace("performing sequence numbers based recovery. starting at [{}]", request.startingSeqNo()); logger.trace("performing sequence numbers based recovery. starting at [{}]", request.startingSeqNo());
startingSeqNo = request.startingSeqNo(); startingSeqNo = request.startingSeqNo();
if (softDeletesEnabled && retentionLeaseRef.get() == null) { if (retentionLeaseRef.get() == null) {
createRetentionLease(startingSeqNo, ActionListener.map(sendFileStep, ignored -> SendFileResult.EMPTY)); createRetentionLease(startingSeqNo, ActionListener.map(sendFileStep, ignored -> SendFileResult.EMPTY));
} else { } else {
sendFileStep.onResponse(SendFileResult.EMPTY); sendFileStep.onResponse(SendFileResult.EMPTY);
@ -253,7 +253,6 @@ public class RecoverySourceHandler {
}); });
final StepListener<ReplicationResponse> deleteRetentionLeaseStep = new StepListener<>(); final StepListener<ReplicationResponse> deleteRetentionLeaseStep = new StepListener<>();
if (softDeletesEnabled) {
runUnderPrimaryPermit(() -> { runUnderPrimaryPermit(() -> {
try { try {
// If the target previously had a copy of this shard then a file-based recovery might move its global // If the target previously had a copy of this shard then a file-based recovery might move its global
@ -266,23 +265,12 @@ public class RecoverySourceHandler {
logger.debug("no peer-recovery retention lease for " + request.targetAllocationId()); logger.debug("no peer-recovery retention lease for " + request.targetAllocationId());
deleteRetentionLeaseStep.onResponse(null); deleteRetentionLeaseStep.onResponse(null);
} }
}, shardId + " removing retention leaes for [" + request.targetAllocationId() + "]", }, shardId + " removing retention lease for [" + request.targetAllocationId() + "]",
shard, cancellableThreads, logger); shard, cancellableThreads, logger);
} else {
deleteRetentionLeaseStep.onResponse(null);
}
deleteRetentionLeaseStep.whenComplete(ignored -> { deleteRetentionLeaseStep.whenComplete(ignored -> {
assert Transports.assertNotTransportThread(RecoverySourceHandler.this + "[phase1]"); assert Transports.assertNotTransportThread(RecoverySourceHandler.this + "[phase1]");
phase1(safeCommitRef.getIndexCommit(), startingSeqNo, () -> estimateNumOps, sendFileStep);
final Consumer<ActionListener<RetentionLease>> createRetentionLeaseAsync;
if (softDeletesEnabled) {
createRetentionLeaseAsync = l -> createRetentionLease(startingSeqNo, l);
} else {
createRetentionLeaseAsync = l -> l.onResponse(null);
}
phase1(safeCommitRef.getIndexCommit(), createRetentionLeaseAsync, () -> estimateNumOps, sendFileStep);
}, onFailure); }, onFailure);
} catch (final Exception e) { } catch (final Exception e) {
@ -454,8 +442,7 @@ public class RecoverySourceHandler {
* segments that are missing. Only segments that have the same size and * segments that are missing. Only segments that have the same size and
* checksum can be reused * checksum can be reused
*/ */
void phase1(IndexCommit snapshot, Consumer<ActionListener<RetentionLease>> createRetentionLease, void phase1(IndexCommit snapshot, long startingSeqNo, IntSupplier translogOps, ActionListener<SendFileResult> listener) {
IntSupplier translogOps, ActionListener<SendFileResult> listener) {
cancellableThreads.checkForCancel(); cancellableThreads.checkForCancel();
final Store store = shard.store(); final Store store = shard.store();
try { try {
@ -529,7 +516,7 @@ public class RecoverySourceHandler {
sendFileInfoStep.whenComplete(r -> sendFileInfoStep.whenComplete(r ->
sendFiles(store, phase1Files.toArray(new StoreFileMetaData[0]), translogOps, sendFilesStep), listener::onFailure); sendFiles(store, phase1Files.toArray(new StoreFileMetaData[0]), translogOps, sendFilesStep), listener::onFailure);
sendFilesStep.whenComplete(r -> createRetentionLease.accept(createRetentionLeaseStep), listener::onFailure); sendFilesStep.whenComplete(r -> createRetentionLease(startingSeqNo, createRetentionLeaseStep), listener::onFailure);
createRetentionLeaseStep.whenComplete(retentionLease -> createRetentionLeaseStep.whenComplete(retentionLease ->
{ {
@ -557,7 +544,7 @@ public class RecoverySourceHandler {
// but we must still create a retention lease // but we must still create a retention lease
final StepListener<RetentionLease> createRetentionLeaseStep = new StepListener<>(); final StepListener<RetentionLease> createRetentionLeaseStep = new StepListener<>();
createRetentionLease.accept(createRetentionLeaseStep); createRetentionLease(startingSeqNo, createRetentionLeaseStep);
createRetentionLeaseStep.whenComplete(retentionLease -> { createRetentionLeaseStep.whenComplete(retentionLease -> {
final TimeValue took = stopWatch.totalTime(); final TimeValue took = stopWatch.totalTime();
logger.trace("recovery [phase1]: took [{}]", took); logger.trace("recovery [phase1]: took [{}]", took);
@ -593,7 +580,8 @@ public class RecoverySourceHandler {
// it's possible that the primary has no retention lease yet if we are doing a rolling upgrade from a version before // it's possible that the primary has no retention lease yet if we are doing a rolling upgrade from a version before
// 7.4, and in that case we just create a lease using the local checkpoint of the safe commit which we're using for // 7.4, and in that case we just create a lease using the local checkpoint of the safe commit which we're using for
// recovery as a conservative estimate for the global checkpoint. // recovery as a conservative estimate for the global checkpoint.
assert shard.indexSettings().getIndexVersionCreated().before(Version.V_7_4_0); assert shard.indexSettings().getIndexVersionCreated().before(Version.V_7_4_0)
|| shard.indexSettings().isSoftDeleteEnabled() == false;
final StepListener<ReplicationResponse> addRetentionLeaseStep = new StepListener<>(); final StepListener<ReplicationResponse> addRetentionLeaseStep = new StepListener<>();
final long estimatedGlobalCheckpoint = startingSeqNo - 1; final long estimatedGlobalCheckpoint = startingSeqNo - 1;
final RetentionLease newLease = shard.addPeerRecoveryRetentionLease(request.targetNode().getId(), final RetentionLease newLease = shard.addPeerRecoveryRetentionLease(request.targetNode().getId(),

8
server/src/test/java/org/elasticsearch/gateway/ReplicaShardAllocatorIT.java
View File

@ -78,7 +78,7 @@ public class ReplicaShardAllocatorIT extends ESIntegTestCase {
assertAcked( assertAcked(
client().admin().indices().prepareCreate(indexName) client().admin().indices().prepareCreate(indexName)
.setSettings(Settings.builder() .setSettings(Settings.builder()
.put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), true) .put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), randomBoolean())
.put(IndexMetaData.SETTING_NUMBER_OF_SHARDS, 1) .put(IndexMetaData.SETTING_NUMBER_OF_SHARDS, 1)
.put(IndexMetaData.SETTING_NUMBER_OF_REPLICAS, 1) .put(IndexMetaData.SETTING_NUMBER_OF_REPLICAS, 1)
.put(IndexSettings.FILE_BASED_RECOVERY_THRESHOLD_SETTING.getKey(), 1.0f) .put(IndexSettings.FILE_BASED_RECOVERY_THRESHOLD_SETTING.getKey(), 1.0f)
@ -211,7 +211,7 @@ public class ReplicaShardAllocatorIT extends ESIntegTestCase {
assertAcked( assertAcked(
client().admin().indices().prepareCreate(indexName) client().admin().indices().prepareCreate(indexName)
.setSettings(Settings.builder() .setSettings(Settings.builder()
.put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), true) .put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), randomBoolean())
.put(IndexMetaData.SETTING_NUMBER_OF_SHARDS, 1) .put(IndexMetaData.SETTING_NUMBER_OF_SHARDS, 1)
.put(IndexSettings.INDEX_TRANSLOG_FLUSH_THRESHOLD_SIZE_SETTING.getKey(), randomIntBetween(10, 100) + "kb") .put(IndexSettings.INDEX_TRANSLOG_FLUSH_THRESHOLD_SIZE_SETTING.getKey(), randomIntBetween(10, 100) + "kb")
.put(IndexMetaData.SETTING_NUMBER_OF_REPLICAS, numOfReplicas) .put(IndexMetaData.SETTING_NUMBER_OF_REPLICAS, numOfReplicas)
@ -248,7 +248,7 @@ public class ReplicaShardAllocatorIT extends ESIntegTestCase {
assertAcked( assertAcked(
client().admin().indices().prepareCreate(indexName) client().admin().indices().prepareCreate(indexName)
.setSettings(Settings.builder() .setSettings(Settings.builder()
.put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), true) .put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), randomBoolean())
.put(IndexMetaData.SETTING_NUMBER_OF_SHARDS, 1) .put(IndexMetaData.SETTING_NUMBER_OF_SHARDS, 1)
.put(IndexSettings.INDEX_TRANSLOG_FLUSH_THRESHOLD_SIZE_SETTING.getKey(), randomIntBetween(10, 100) + "kb") .put(IndexSettings.INDEX_TRANSLOG_FLUSH_THRESHOLD_SIZE_SETTING.getKey(), randomIntBetween(10, 100) + "kb")
.put(IndexMetaData.SETTING_NUMBER_OF_REPLICAS, 1) .put(IndexMetaData.SETTING_NUMBER_OF_REPLICAS, 1)
@ -329,7 +329,7 @@ public class ReplicaShardAllocatorIT extends ESIntegTestCase {
createIndex(indexName, Settings.builder() createIndex(indexName, Settings.builder()
.put(IndexMetaData.SETTING_NUMBER_OF_SHARDS, 1) .put(IndexMetaData.SETTING_NUMBER_OF_SHARDS, 1)
.put(IndexMetaData.SETTING_NUMBER_OF_REPLICAS, 0) .put(IndexMetaData.SETTING_NUMBER_OF_REPLICAS, 0)
.put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), true) .put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), randomBoolean())
.put(IndexService.GLOBAL_CHECKPOINT_SYNC_INTERVAL_SETTING.getKey(), "100ms") .put(IndexService.GLOBAL_CHECKPOINT_SYNC_INTERVAL_SETTING.getKey(), "100ms")
.put(IndexService.RETENTION_LEASE_SYNC_INTERVAL_SETTING.getKey(), "100ms") .put(IndexService.RETENTION_LEASE_SYNC_INTERVAL_SETTING.getKey(), "100ms")
.build()); .build());

30
server/src/test/java/org/elasticsearch/index/seqno/RetentionLeaseIT.java
View File

@ -335,36 +335,6 @@ public class RetentionLeaseIT extends ESIntegTestCase {
} }
} }
public void testRetentionLeasesBackgroundSyncWithSoftDeletesDisabled() throws Exception {
final int numberOfReplicas = 2 - scaledRandomIntBetween(0, 2);
internalCluster().ensureAtLeastNumDataNodes(1 + numberOfReplicas);
TimeValue syncIntervalSetting = TimeValue.timeValueMillis(between(1, 100));
final Settings settings = Settings.builder()
.put("index.number_of_shards", 1)
.put("index.number_of_replicas", numberOfReplicas)
.put(IndexService.RETENTION_LEASE_SYNC_INTERVAL_SETTING.getKey(), syncIntervalSetting.getStringRep())
.put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), false)
.build();
createIndex("index", settings);
final String primaryShardNodeId = clusterService().state().routingTable().index("index").shard(0).primaryShard().currentNodeId();
final String primaryShardNodeName = clusterService().state().nodes().get(primaryShardNodeId).getName();
final MockTransportService primaryTransportService = (MockTransportService) internalCluster().getInstance(
TransportService.class, primaryShardNodeName);
final AtomicBoolean backgroundSyncRequestSent = new AtomicBoolean();
primaryTransportService.addSendBehavior((connection, requestId, action, request, options) -> {
if (action.startsWith(RetentionLeaseBackgroundSyncAction.ACTION_NAME)) {
backgroundSyncRequestSent.set(true);
}
connection.sendRequest(requestId, action, request, options);
});
final long start = System.nanoTime();
ensureGreen("index");
final long syncEnd = System.nanoTime();
// We sleep long enough for the retention leases background sync to be triggered
Thread.sleep(Math.max(0, randomIntBetween(2, 3) * syncIntervalSetting.millis() - TimeUnit.NANOSECONDS.toMillis(syncEnd - start)));
assertFalse("retention leases background sync must be a noop if soft deletes is disabled", backgroundSyncRequestSent.get());
}
public void testRetentionLeasesSyncOnRecovery() throws Exception { public void testRetentionLeasesSyncOnRecovery() throws Exception {
final int numberOfReplicas = 2 - scaledRandomIntBetween(0, 2); final int numberOfReplicas = 2 - scaledRandomIntBetween(0, 2);
internalCluster().ensureAtLeastNumDataNodes(1 + numberOfReplicas); internalCluster().ensureAtLeastNumDataNodes(1 + numberOfReplicas);

3
server/src/test/java/org/elasticsearch/index/shard/IndexShardRetentionLeaseTests.java
View File

@ -314,8 +314,7 @@ public class IndexShardRetentionLeaseTests extends IndexShardTestCase {
assertThat(expectThrows(AssertionError.class, () -> shard.removeRetentionLease( assertThat(expectThrows(AssertionError.class, () -> shard.removeRetentionLease(
randomAlphaOfLength(10), ActionListener.wrap(() -> {}))).getMessage(), randomAlphaOfLength(10), ActionListener.wrap(() -> {}))).getMessage(),
equalTo("retention leases requires soft deletes but [index] does not have soft deletes enabled")); equalTo("retention leases requires soft deletes but [index] does not have soft deletes enabled"));
assertThat(expectThrows(AssertionError.class, shard::syncRetentionLeases).getMessage(), shard.syncRetentionLeases();
equalTo("retention leases requires soft deletes but [index] does not have soft deletes enabled"));
closeShards(shard); closeShards(shard);
} }

9
server/src/test/java/org/elasticsearch/indices/recovery/RecoverySourceHandlerTests.java
View File

@ -64,7 +64,6 @@ import org.elasticsearch.index.mapper.ParsedDocument;
import org.elasticsearch.index.mapper.SeqNoFieldMapper; import org.elasticsearch.index.mapper.SeqNoFieldMapper;
import org.elasticsearch.index.mapper.Uid; import org.elasticsearch.index.mapper.Uid;
import org.elasticsearch.index.seqno.ReplicationTracker; import org.elasticsearch.index.seqno.ReplicationTracker;
import org.elasticsearch.index.seqno.RetentionLease;
import org.elasticsearch.index.seqno.RetentionLeases; import org.elasticsearch.index.seqno.RetentionLeases;
import org.elasticsearch.index.seqno.SeqNoStats; import org.elasticsearch.index.seqno.SeqNoStats;
import org.elasticsearch.index.seqno.SequenceNumbers; import org.elasticsearch.index.seqno.SequenceNumbers;
@ -102,7 +101,6 @@ import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;
import java.util.function.IntSupplier; import java.util.function.IntSupplier;
import java.util.zip.CRC32; import java.util.zip.CRC32;
@ -468,10 +466,9 @@ public class RecoverySourceHandlerTests extends ESTestCase {
between(1, 8)) { between(1, 8)) {
@Override @Override
void phase1(IndexCommit snapshot, Consumer<ActionListener<RetentionLease>> createRetentionLease, void phase1(IndexCommit snapshot, long startingSeqNo, IntSupplier translogOps, ActionListener<SendFileResult> listener) {
IntSupplier translogOps, ActionListener<SendFileResult> listener) {
phase1Called.set(true); phase1Called.set(true);
super.phase1(snapshot, createRetentionLease, translogOps, listener); super.phase1(snapshot, startingSeqNo, translogOps, listener);
} }
@Override @Override
@ -687,7 +684,7 @@ public class RecoverySourceHandlerTests extends ESTestCase {
try { try {
final CountDownLatch latch = new CountDownLatch(1); final CountDownLatch latch = new CountDownLatch(1);
handler.phase1(DirectoryReader.listCommits(dir).get(0), handler.phase1(DirectoryReader.listCommits(dir).get(0),
l -> recoveryExecutor.execute(() -> l.onResponse(null)), 0,
() -> 0, () -> 0,
new LatchedActionListener<>(phase1Listener, latch)); new LatchedActionListener<>(phase1Listener, latch));
latch.await(); latch.await();

44
test/framework/src/main/java/org/elasticsearch/test/rest/ESRestTestCase.java
View File

@ -52,9 +52,11 @@ import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.common.xcontent.json.JsonXContent; import org.elasticsearch.common.xcontent.json.JsonXContent;
import org.elasticsearch.common.xcontent.support.XContentMapValues; import org.elasticsearch.common.xcontent.support.XContentMapValues;
import org.elasticsearch.core.internal.io.IOUtils; import org.elasticsearch.core.internal.io.IOUtils;
import org.elasticsearch.index.seqno.ReplicationTracker;
import org.elasticsearch.rest.RestStatus; import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.snapshots.SnapshotState; import org.elasticsearch.snapshots.SnapshotState;
import org.elasticsearch.test.ESTestCase; import org.elasticsearch.test.ESTestCase;
import org.elasticsearch.test.rest.yaml.ObjectPath;
import org.hamcrest.Matchers; import org.hamcrest.Matchers;
import org.junit.After; import org.junit.After;
import org.junit.AfterClass; import org.junit.AfterClass;
@ -86,12 +88,15 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer; import java.util.function.Consumer;
import java.util.function.Predicate; import java.util.function.Predicate;
import java.util.stream.Collectors;
import static java.util.Collections.sort; import static java.util.Collections.sort;
import static java.util.Collections.unmodifiableList; import static java.util.Collections.unmodifiableList;
import static org.hamcrest.Matchers.anEmptyMap; import static org.hamcrest.Matchers.anEmptyMap;
import static org.hamcrest.Matchers.anyOf; import static org.hamcrest.Matchers.anyOf;
import static org.hamcrest.Matchers.equalTo; import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.everyItem;
import static org.hamcrest.Matchers.in;
/** /**
* Superclass for tests that interact with an external test cluster using Elasticsearch's {@link RestClient}. * Superclass for tests that interact with an external test cluster using Elasticsearch's {@link RestClient}.
@ -1134,6 +1139,7 @@ public abstract class ESRestTestCase extends ESTestCase {
* that we have renewed every PRRL to the global checkpoint of the corresponding copy and properly synced to all copies. * that we have renewed every PRRL to the global checkpoint of the corresponding copy and properly synced to all copies.
*/ */
public void ensurePeerRecoveryRetentionLeasesRenewedAndSynced(String index) throws Exception { public void ensurePeerRecoveryRetentionLeasesRenewedAndSynced(String index) throws Exception {
final boolean alwaysExists = minimumNodeVersion().onOrAfter(Version.V_7_6_0);
assertBusy(() -> { assertBusy(() -> {
Map<String, Object> stats = entityAsMap(client().performRequest(new Request("GET", index + "/_stats?level=shards"))); Map<String, Object> stats = entityAsMap(client().performRequest(new Request("GET", index + "/_stats?level=shards")));
@SuppressWarnings("unchecked") Map<String, List<Map<String, ?>>> shards = @SuppressWarnings("unchecked") Map<String, List<Map<String, ?>>> shards =
@ -1145,16 +1151,52 @@ public abstract class ESRestTestCase extends ESTestCase {
assertThat(XContentMapValues.extractValue("seq_no.max_seq_no", copy), equalTo(globalCheckpoint)); assertThat(XContentMapValues.extractValue("seq_no.max_seq_no", copy), equalTo(globalCheckpoint));
@SuppressWarnings("unchecked") List<Map<String, ?>> retentionLeases = @SuppressWarnings("unchecked") List<Map<String, ?>> retentionLeases =
(List<Map<String, ?>>) XContentMapValues.extractValue("retention_leases.leases", copy); (List<Map<String, ?>>) XContentMapValues.extractValue("retention_leases.leases", copy);
if (retentionLeases == null) { if (alwaysExists == false && retentionLeases == null) {
continue; continue;
} }
assertNotNull(retentionLeases);
for (Map<String, ?> retentionLease : retentionLeases) { for (Map<String, ?> retentionLease : retentionLeases) {
if (((String) retentionLease.get("id")).startsWith("peer_recovery/")) { if (((String) retentionLease.get("id")).startsWith("peer_recovery/")) {
assertThat(retentionLease.get("retaining_seq_no"), equalTo(globalCheckpoint + 1)); assertThat(retentionLease.get("retaining_seq_no"), equalTo(globalCheckpoint + 1));
} }
} }
if (alwaysExists) {
List<String> existingLeaseIds = retentionLeases.stream().map(lease -> (String) lease.get("id"))
.collect(Collectors.toList());
List<String> expectedLeaseIds = shard.stream()
.map(shr -> (String) XContentMapValues.extractValue("routing.node", shr))
.map(ReplicationTracker::getPeerRecoveryRetentionLeaseId)
.collect(Collectors.toList());
assertThat("not every active copy has established its PPRL", expectedLeaseIds, everyItem(in(existingLeaseIds)));
}
} }
} }
}, 60, TimeUnit.SECONDS); }, 60, TimeUnit.SECONDS);
} }
public static Boolean getHasXPack() {
return hasXPack;
}
/**
* Returns the minimum node version among all nodes of the cluster
*/
protected static Version minimumNodeVersion() throws IOException {
final Request request = new Request("GET", "_nodes");
request.addParameter("filter_path", "nodes.*.version");
final Response response = client().performRequest(request);
final Map<String, Object> nodes = ObjectPath.createFromResponse(response).evaluate("nodes");
Version minVersion = null;
for (Map.Entry<String, Object> node : nodes.entrySet()) {
@SuppressWarnings("unchecked")
Version nodeVersion = Version.fromString((String) ((Map<String, Object>) node.getValue()).get("version"));
if (minVersion == null || minVersion.after(nodeVersion)) {
minVersion = nodeVersion;
}
}
assertNotNull(minVersion);
return minVersion;
}
} }