-
Notifications
You must be signed in to change notification settings - Fork 1.8k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[Backport 2.x] [Segment Replication] Refactor file cleanup logic and …
…fix PIT/Scroll with remote store. (#9272) * [Segment Replication] Refactor file cleanup logic and fix PIT/Scroll with remote store. (#9111) * Remove divergent commit logic with segment replication. This change removes divergent commit paths for segrep node-node and remote store. All replicas with segrep enabled will perform local commits and ignore any incoming segments_n file. This changes the recovery sync with remote store to also exclude the segments_n so that only the fetched infos bytes are committed before an engine is opened. This change also updates deletion logic with segment replication to automatically delete when a file is decref'd to 0. Signed-off-by: Marc Handalian <handalm@amazon.com> * Add more NRTReplicationEngineTests. Signed-off-by: Marc Handalian <handalm@amazon.com> * Ensure old commit files are wiped on remote store sync before we commit a new segmentInfos. Signed-off-by: Marc Handalian <handalm@amazon.com> * Add more shard level tests. Signed-off-by: Marc Handalian <handalm@amazon.com> * Add test ensuring commits are cleaned up on replicas. Signed-off-by: Marc Handalian <handalm@amazon.com> * Self review. Signed-off-by: Marc Handalian <handalm@amazon.com> * Use refresh level sync before recovery Signed-off-by: Marc Handalian <handalm@amazon.com> * PR feedback. Signed-off-by: Marc Handalian <handalm@amazon.com> --------- Signed-off-by: Marc Handalian <handalm@amazon.com> (cherry picked from commit c301544) * Fix test SegmentReplicationIndexShardTests.testPrimaryRestart. This test is specific to remote store and should not be run for node-node replication. Signed-off-by: Marc Handalian <handalm@amazon.com> (cherry picked from commit a33f67e) Signed-off-by: Marc Handalian <handalm@amazon.com> --------- Signed-off-by: Marc Handalian <handalm@amazon.com>
- Loading branch information
Showing
17 changed files
with
595 additions
and
349 deletions.
There are no files selected for viewing
This file contains 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 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 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
92 changes: 92 additions & 0 deletions
92
server/src/main/java/org/opensearch/index/engine/ReplicaFileTracker.java
This file contains 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,92 @@ | ||
| /* | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| * | ||
| * The OpenSearch Contributors require contributions made to | ||
| * this file be licensed under the Apache-2.0 license or a | ||
| * compatible open source license. | ||
| */ | ||
|
|
||
| package org.opensearch.index.engine; | ||
|
|
||
| import org.apache.logging.log4j.LogManager; | ||
| import org.apache.logging.log4j.Logger; | ||
|
|
||
| import java.util.Collection; | ||
| import java.util.HashMap; | ||
| import java.util.HashSet; | ||
| import java.util.Map; | ||
| import java.util.Optional; | ||
| import java.util.Set; | ||
| import java.util.function.BiConsumer; | ||
|
|
||
| /** | ||
| * This class is heavily influenced by Lucene's ReplicaFileDeleter class used to keep track of | ||
| * segment files that should be preserved on replicas between replication events. | ||
| * | ||
| * https://github.com/apache/lucene/blob/main/lucene/replicator/src/java/org/apache/lucene/replicator/nrt/ReplicaFileDeleter.java | ||
| * | ||
| * @opensearch.internal | ||
| */ | ||
| final class ReplicaFileTracker { | ||
|
|
||
| public static final Logger logger = LogManager.getLogger(ReplicaFileTracker.class); | ||
| private final Map<String, Integer> refCounts = new HashMap<>(); | ||
| private final BiConsumer<String, String> fileDeleter; | ||
| private final Set<String> EXCLUDE_FILES = Set.of("write.lock"); | ||
|
|
||
| public ReplicaFileTracker(BiConsumer<String, String> fileDeleter) { | ||
| this.fileDeleter = fileDeleter; | ||
| } | ||
|
|
||
| public synchronized void incRef(Collection<String> fileNames) { | ||
| for (String fileName : fileNames) { | ||
| refCounts.merge(fileName, 1, Integer::sum); | ||
| } | ||
| } | ||
|
|
||
| public synchronized int refCount(String file) { | ||
| return Optional.ofNullable(refCounts.get(file)).orElse(0); | ||
| } | ||
|
|
||
| public synchronized void decRef(Collection<String> fileNames) { | ||
| Set<String> toDelete = new HashSet<>(); | ||
| for (String fileName : fileNames) { | ||
| Integer curCount = refCounts.get(fileName); | ||
| assert curCount != null : "fileName=" + fileName; | ||
| assert curCount > 0; | ||
| if (curCount == 1) { | ||
| refCounts.remove(fileName); | ||
| toDelete.add(fileName); | ||
| } else { | ||
| refCounts.put(fileName, curCount - 1); | ||
| } | ||
| } | ||
| if (toDelete.isEmpty() == false) { | ||
| delete(toDelete); | ||
| } | ||
| } | ||
|
|
||
| public void deleteUnreferencedFiles(String... toDelete) { | ||
| for (String file : toDelete) { | ||
| if (canDelete(file)) { | ||
| delete(file); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private synchronized void delete(Collection<String> toDelete) { | ||
| for (String fileName : toDelete) { | ||
| delete(fileName); | ||
| } | ||
| } | ||
|
|
||
| private synchronized void delete(String fileName) { | ||
| assert canDelete(fileName); | ||
| fileDeleter.accept("delete unreferenced", fileName); | ||
| } | ||
|
|
||
| private synchronized boolean canDelete(String fileName) { | ||
| return EXCLUDE_FILES.contains(fileName) == false && refCounts.containsKey(fileName) == false; | ||
| } | ||
|
|
||
| } |
Oops, something went wrong.