Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Handle replication cases where the remote store metadata file is not present #8433

Merged
merged 3 commits into from
Jul 14, 2023
Merged

Handle replication cases where the remote store metadata file is not present #8433

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
bd25364
Handle replication cases where the remote store metadata file is not …
ankitkala Jun 29, 2023
041e614
Allow null metadata file only if shard is not started
ankitkala Jul 7, 2023
0191916
Merge branch 'main' into no_metadata
ankitkala Jul 11, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 10 additions & 2 deletions server/src/main/java/org/opensearch/indices/replication/RemoteStoreReplicationSource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,9 +14,11 @@
import org.apache.lucene.util.Version;
import org.opensearch.action.ActionListener;
import org.opensearch.index.shard.IndexShard;
import org.opensearch.index.shard.IndexShardState;
import org.opensearch.index.store.RemoteSegmentStoreDirectory;
import org.opensearch.index.store.Store;
import org.opensearch.index.store.StoreFileMetadata;
import org.opensearch.index.store.remote.metadata.RemoteSegmentMetadata;
import org.opensearch.indices.replication.checkpoint.ReplicationCheckpoint;

import java.util.Collections;
Expand Down Expand Up @@ -53,8 +55,14 @@ public void getCheckpointMetadata(
// TODO: Need to figure out a way to pass this information for segment metadata via remote store.
final Version version = indexShard.getSegmentInfosSnapshot().get().getCommitLuceneVersion();
try {
metadataMap = remoteDirectory.readLatestMetadataFile()
.getMetadata()
RemoteSegmentMetadata mdFile = remoteDirectory.readLatestMetadataFile();
// During initial recovery flow, the remote store might not have metadata as primary hasn't uploaded anything yet.
if (mdFile == null && indexShard.state().equals(IndexShardState.STARTED) == false) {
mch2 marked this conversation as resolved.
Show resolved Hide resolved
listener.onResponse(new CheckpointInfoResponse(checkpoint, Collections.emptyMap(), null));
return;
}
assert mdFile != null : "Remote metadata file can't be null if shard is active " + indexShard.state();
metadataMap = mdFile.getMetadata()
.entrySet()
.stream()
.collect(
Expand Down
49 changes: 47 additions & 2 deletions ...r/src/test/java/org/opensearch/indices/replication/RemoteStoreReplicationSourceTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
import org.opensearch.index.engine.InternalEngineFactory;
import org.opensearch.index.replication.OpenSearchIndexLevelReplicationTestCase;
import org.opensearch.index.shard.IndexShard;
import org.opensearch.index.shard.IndexShardState;
import org.opensearch.index.shard.RemoteStoreRefreshListenerTests;
import org.opensearch.index.store.RemoteSegmentStoreDirectory;
import org.opensearch.index.store.Store;
Expand All @@ -41,6 +42,8 @@ public class RemoteStoreReplicationSourceTests extends OpenSearchIndexLevelRepli

private IndexShard mockShard;

private Store remoteStore;

@Override
public void setUp() throws Exception {
super.setUp();
Expand All @@ -59,7 +62,7 @@ public void setUp() throws Exception {
Store store = mock(Store.class);
when(mockShard.store()).thenReturn(store);
when(store.directory()).thenReturn(indexShard.store().directory());
Store remoteStore = mock(Store.class);
remoteStore = mock(Store.class);
when(mockShard.remoteStore()).thenReturn(remoteStore);
RemoteSegmentStoreDirectory remoteSegmentStoreDirectory =
(RemoteSegmentStoreDirectory) ((FilterDirectory) ((FilterDirectory) indexShard.remoteStore().directory()).getDelegate())
Expand All @@ -68,7 +71,6 @@ public void setUp() throws Exception {
new RemoteStoreRefreshListenerTests.TestFilterDirectory(remoteSegmentStoreDirectory)
);
when(remoteStore.directory()).thenReturn(remoteStoreFilterDirectory);

replicationSource = new RemoteStoreReplicationSource(mockShard);
}

Expand Down Expand Up @@ -113,6 +115,49 @@ public void testGetCheckpointMetadataFailure() {
});
}

public void testGetCheckpointMetadataEmpty() throws ExecutionException, InterruptedException, IOException {
when(mockShard.getSegmentInfosSnapshot()).thenReturn(indexShard.getSegmentInfosSnapshot());
final ReplicationCheckpoint checkpoint = new ReplicationCheckpoint(
indexShard.shardId(),
PRIMARY_TERM,
SEGMENTS_GEN,
VERSION,
Codec.getDefault().getName()
);
IndexShard emptyIndexShard = null;
try {
emptyIndexShard = newStartedShard(
true,
Settings.builder().put(IndexMetadata.SETTING_REMOTE_STORE_ENABLED, true).build(),
new InternalEngineFactory()
);
RemoteSegmentStoreDirectory remoteSegmentStoreDirectory =
(RemoteSegmentStoreDirectory) ((FilterDirectory) ((FilterDirectory) emptyIndexShard.remoteStore().directory()).getDelegate())
.getDelegate();
FilterDirectory remoteStoreFilterDirectory = new RemoteStoreRefreshListenerTests.TestFilterDirectory(
new RemoteStoreRefreshListenerTests.TestFilterDirectory(remoteSegmentStoreDirectory)
);
when(remoteStore.directory()).thenReturn(remoteStoreFilterDirectory);

final PlainActionFuture<CheckpointInfoResponse> res = PlainActionFuture.newFuture();
when(mockShard.state()).thenReturn(IndexShardState.RECOVERING);
// Recovering shard should just do a noop and return empty metadata map.
replicationSource.getCheckpointMetadata(REPLICATION_ID, checkpoint, res);
CheckpointInfoResponse response = res.get();
assert (response.getCheckpoint().equals(checkpoint));
assert (response.getMetadataMap().isEmpty());

when(mockShard.state()).thenReturn(IndexShardState.STARTED);
// Started shard should fail with assertion error.
expectThrows(AssertionError.class, () -> {
final PlainActionFuture<CheckpointInfoResponse> res2 = PlainActionFuture.newFuture();
replicationSource.getCheckpointMetadata(REPLICATION_ID, checkpoint, res2);
});
} finally {
closeShards(emptyIndexShard);
}
}

public void testGetSegmentFiles() throws ExecutionException, InterruptedException {
final ReplicationCheckpoint checkpoint = new ReplicationCheckpoint(
indexShard.shardId(),
Expand Down