Skip to content

Commit

Permalink
Merge pull request opensearch-project#151 from Bukhtawar/remote-store…
Browse files Browse the repository at this point in the history 
…-enabled-its

Empty replication leases on replica
  • Loading branch information
@Bukhtawar
Bukhtawar authored Sep 2, 2023
2 parents 27015e4 + 09e9ffa commit 27e1d96
Show file tree
Hide file tree
Showing 2 changed files with 59 additions and 18 deletions.
70 changes: 52 additions & 18 deletions server/src/internalClusterTest/java/org/opensearch/index/seqno/RetentionLeaseIT.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@
import org.opensearch.common.settings.Settings;
import org.opensearch.common.unit.TimeValue;
import org.opensearch.core.action.ActionListener;
import org.opensearch.core.common.util.CollectionUtils;
import org.opensearch.core.index.shard.ShardId;
import org.opensearch.index.IndexService;
import org.opensearch.index.IndexSettings;
Expand Down Expand Up @@ -70,12 +71,12 @@
import java.util.stream.Collectors;
import java.util.stream.Stream;

import static org.opensearch.indices.recovery.RecoverySettings.INDICES_RECOVERY_RETRY_DELAY_NETWORK_SETTING;
import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked;
import static org.hamcrest.Matchers.anyOf;
import static org.hamcrest.Matchers.contains;
import static org.hamcrest.Matchers.empty;
import static org.hamcrest.Matchers.equalTo;
import static org.opensearch.indices.recovery.RecoverySettings.INDICES_RECOVERY_RETRY_DELAY_NETWORK_SETTING;
import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked;

@OpenSearchIntegTestCase.ClusterScope(scope = OpenSearchIntegTestCase.Scope.TEST)
public class RetentionLeaseIT extends OpenSearchIntegTestCase {
Expand Down Expand Up @@ -141,13 +142,22 @@ public void testRetentionLeasesSyncedOnAdd() throws Exception {
final Map<String, RetentionLease> retentionLeasesOnReplica = RetentionLeaseUtils.toMapExcludingPeerRecoveryRetentionLeases(
replica.getRetentionLeases()
);
assertThat(retentionLeasesOnReplica, equalTo(currentRetentionLeases));
if (isIndexRemoteStoreEnabled("index")) {
assertThat(retentionLeasesOnReplica, equalTo(Collections.EMPTY_MAP));
} else {
assertThat(retentionLeasesOnReplica, equalTo(currentRetentionLeases));
}

// check retention leases have been written on the replica
assertThat(
currentRetentionLeases,
equalTo(RetentionLeaseUtils.toMapExcludingPeerRecoveryRetentionLeases(replica.loadRetentionLeases()))
);
if (isIndexRemoteStoreEnabled("index")) {
assertThat(
RetentionLeaseUtils.toMapExcludingPeerRecoveryRetentionLeases(replica.loadRetentionLeases()), equalTo(Collections.EMPTY_MAP)
);
} else {
assertThat(
RetentionLeaseUtils.toMapExcludingPeerRecoveryRetentionLeases(replica.loadRetentionLeases()), equalTo(currentRetentionLeases)
);
}
}
}
}
Expand Down Expand Up @@ -205,13 +215,22 @@ public void testRetentionLeaseSyncedOnRemove() throws Exception {
final Map<String, RetentionLease> retentionLeasesOnReplica = RetentionLeaseUtils.toMapExcludingPeerRecoveryRetentionLeases(
replica.getRetentionLeases()
);
assertThat(retentionLeasesOnReplica, equalTo(currentRetentionLeases));
if (isIndexRemoteStoreEnabled("index")) {
assertThat(retentionLeasesOnReplica, equalTo(Collections.EMPTY_MAP));
} else {
assertThat(retentionLeasesOnReplica, equalTo(currentRetentionLeases));
}

// check retention leases have been written on the replica
assertThat(
currentRetentionLeases,
equalTo(RetentionLeaseUtils.toMapExcludingPeerRecoveryRetentionLeases(replica.loadRetentionLeases()))
);
if (isIndexRemoteStoreEnabled("index")) {
assertThat(
RetentionLeaseUtils.toMapExcludingPeerRecoveryRetentionLeases(replica.loadRetentionLeases()), equalTo(Collections.EMPTY_MAP)
);
} else {
assertThat(
RetentionLeaseUtils.toMapExcludingPeerRecoveryRetentionLeases(replica.loadRetentionLeases()), equalTo(currentRetentionLeases)
);
}
}
}
}
Expand Down Expand Up @@ -352,7 +371,11 @@ public void testBackgroundRetentionLeaseSync() throws Exception {
final String replicaShardNodeName = clusterService().state().nodes().get(replicaShardNodeId).getName();
final IndexShard replica = internalCluster().getInstance(IndicesService.class, replicaShardNodeName)
.getShardOrNull(new ShardId(resolveIndex("index"), 0));
assertThat(replica.getRetentionLeases(), equalTo(primary.getRetentionLeases()));
if(isIndexRemoteStoreEnabled("index")) {
assertThat(replica.getRetentionLeases(), equalTo(new RetentionLeases(primary.getOperationPrimaryTerm(), 0, new ArrayList<>())));
} else {
assertThat(replica.getRetentionLeases(), equalTo(primary.getRetentionLeases()));
}
}
});
}
Expand Down Expand Up @@ -444,13 +467,24 @@ public void testRetentionLeasesSyncOnRecovery() throws Exception {
final Map<String, RetentionLease> retentionLeasesOnReplica = RetentionLeaseUtils.toMapExcludingPeerRecoveryRetentionLeases(
replica.getRetentionLeases()
);
assertThat(retentionLeasesOnReplica, equalTo(currentRetentionLeases));
if(isIndexRemoteStoreEnabled("index")) {
assertThat(retentionLeasesOnReplica, equalTo(Collections.EMPTY_MAP));
} else {
assertThat(retentionLeasesOnReplica, equalTo(currentRetentionLeases));
}

// check retention leases have been written on the replica; see RecoveryTarget#finalizeRecovery
assertThat(
currentRetentionLeases,
equalTo(RetentionLeaseUtils.toMapExcludingPeerRecoveryRetentionLeases(replica.loadRetentionLeases()))
);
if(isIndexRemoteStoreEnabled("index")) {
assertThat(
RetentionLeaseUtils.toMapExcludingPeerRecoveryRetentionLeases(replica.loadRetentionLeases()),
equalTo(Collections.EMPTY_MAP)
);
} else {
assertThat(
RetentionLeaseUtils.toMapExcludingPeerRecoveryRetentionLeases(replica.loadRetentionLeases()),
equalTo(currentRetentionLeases)
);
}
}
}

Expand Down
7 changes: 7 additions & 0 deletions test/framework/src/main/java/org/opensearch/test/OpenSearchIntegTestCase.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,7 @@
import org.opensearch.action.admin.indices.segments.IndexShardSegments;
import org.opensearch.action.admin.indices.segments.IndicesSegmentResponse;
import org.opensearch.action.admin.indices.segments.ShardSegments;
import org.opensearch.action.admin.indices.settings.get.GetSettingsRequest;
import org.opensearch.action.admin.indices.template.put.PutIndexTemplateRequestBuilder;
import org.opensearch.action.bulk.BulkRequestBuilder;
import org.opensearch.action.bulk.BulkResponse;
Expand Down Expand Up @@ -203,6 +204,7 @@
import java.util.concurrent.Callable;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
Expand Down Expand Up @@ -2598,4 +2600,9 @@ protected ClusterState getClusterState() {
return client(internalCluster().getClusterManagerName()).admin().cluster().prepareState().get().getState();
}

protected boolean isIndexRemoteStoreEnabled(String index) throws Exception {
return client().admin().indices().getSettings(new GetSettingsRequest().indices("index")).get()
.getSetting("index", IndexMetadata.SETTING_REMOTE_STORE_ENABLED).equals(Boolean.TRUE.toString());
}

}

0 comments on commit 27e1d96

Please sign in to comment.