Refactor and rename translog manager implementations

Signed-off-by: Varun Bharadwaj <[email protected]>

server/src/main/java/org/opensearch/index/engine/IngestionEngine.java

@@ -22,10 +22,10 @@
 import org.opensearch.index.mapper.IdFieldMapper;
 import org.opensearch.index.mapper.ParseContext;
 import org.opensearch.index.seqno.SequenceNumbers;
-import org.opensearch.index.translog.EngineTranslogManager;
-import org.opensearch.index.translog.NoOpInternalTranslogManager;
+import org.opensearch.index.translog.NoOpEngineTranslogManager;
 import org.opensearch.index.translog.Translog;
 import org.opensearch.index.translog.TranslogDeletionPolicy;
+import org.opensearch.index.translog.TranslogManager;
 import org.opensearch.index.translog.listener.CompositeTranslogEventListener;
 import org.opensearch.indices.pollingingest.DefaultStreamPoller;
 import org.opensearch.indices.pollingingest.StreamPoller;
@@ -268,12 +268,12 @@ public DocumentMapperForType getDocumentMapperForType() {
     }
 
     @Override
-    protected EngineTranslogManager createTranslogManager(
+    protected TranslogManager createTranslogManager(
         String translogUUID,
         TranslogDeletionPolicy translogDeletionPolicy,
         CompositeTranslogEventListener translogEventListener
     ) {
-        return new NoOpInternalTranslogManager(engineConfig, translogUUID);
+        return new NoOpEngineTranslogManager(engineConfig, translogUUID);
     }
 
     protected Map<String, String> commitDataAsMap() {

server/src/main/java/org/opensearch/index/engine/InternalEngine.java

@@ -104,8 +104,7 @@
 import org.opensearch.index.seqno.SeqNoStats;
 import org.opensearch.index.seqno.SequenceNumbers;
 import org.opensearch.index.shard.OpenSearchMergePolicy;
-import org.opensearch.index.translog.EngineTranslogManager;
-import org.opensearch.index.translog.InternalTranslogManager;
+import org.opensearch.index.translog.InternalEngineTranslogManager;
 import org.opensearch.index.translog.Translog;
 import org.opensearch.index.translog.TranslogCorruptedException;
 import org.opensearch.index.translog.TranslogDeletionPolicy;
@@ -156,7 +155,7 @@ public class InternalEngine extends Engine {
      */
     private volatile long lastDeleteVersionPruneTimeMSec;
 
-    protected final EngineTranslogManager translogManager;
+    protected final TranslogManager translogManager;
     protected final IndexWriter indexWriter;
     protected final LocalCheckpointTracker localCheckpointTracker;
     protected final AtomicLong maxUnsafeAutoIdTimestamp = new AtomicLong(-1);
@@ -248,7 +247,7 @@ public TranslogManager translogManager() {
         ExternalReaderManager externalReaderManager = null;
         OpenSearchReaderManager internalReaderManager = null;
         EngineMergeScheduler scheduler = null;
-        EngineTranslogManager translogManagerRef = null;
+        TranslogManager translogManagerRef = null;
         boolean success = false;
         try {
             this.lastDeleteVersionPruneTimeMSec = engineConfig.getThreadPool().relativeTimeInMillis();
@@ -352,12 +351,12 @@ public void onFailure(String reason, Exception ex) {
         logger.trace("created new InternalEngine");
     }
 
-    protected EngineTranslogManager createTranslogManager(
+    protected TranslogManager createTranslogManager(
         String translogUUID,
         TranslogDeletionPolicy translogDeletionPolicy,
         CompositeTranslogEventListener translogEventListener
     ) throws IOException {
-        return new InternalTranslogManager(
+        return new InternalEngineTranslogManager(
             engineConfig.getTranslogConfig(),
             engineConfig.getPrimaryTermSupplier(),
             engineConfig.getGlobalCheckpointSupplier(),

server/src/main/java/org/opensearch/index/engine/NRTReplicationEngine.java

@@ -29,7 +29,7 @@
 import org.opensearch.index.translog.TranslogDeletionPolicy;
 import org.opensearch.index.translog.TranslogException;
 import org.opensearch.index.translog.TranslogManager;
-import org.opensearch.index.translog.WriteOnlyTranslogManager;
+import org.opensearch.index.translog.WriteOnlyInternalEngineTranslogManager;
 import org.opensearch.index.translog.listener.TranslogEventListener;
 import org.opensearch.search.suggest.completion.CompletionStats;
 
@@ -61,7 +61,7 @@ public class NRTReplicationEngine extends Engine {
     private final NRTReplicationReaderManager readerManager;
     private final CompletionStatsCache completionStatsCache;
     private final LocalCheckpointTracker localCheckpointTracker;
-    private final WriteOnlyTranslogManager translogManager;
+    private final WriteOnlyInternalEngineTranslogManager translogManager;
     private final Lock flushLock = new ReentrantLock();
     protected final ReplicaFileTracker replicaFileTracker;
 
@@ -73,7 +73,7 @@ public NRTReplicationEngine(EngineConfig engineConfig) {
         super(engineConfig);
         store.incRef();
         NRTReplicationReaderManager readerManager = null;
-        WriteOnlyTranslogManager translogManagerRef = null;
+        WriteOnlyInternalEngineTranslogManager translogManagerRef = null;
         boolean success = false;
         try {
             this.replicaFileTracker = new ReplicaFileTracker(store::deleteQuiet);
@@ -99,7 +99,7 @@ public NRTReplicationEngine(EngineConfig engineConfig) {
             }
             final Map<String, String> userData = this.lastCommittedSegmentInfos.getUserData();
             final String translogUUID = Objects.requireNonNull(userData.get(Translog.TRANSLOG_UUID_KEY));
-            translogManagerRef = new WriteOnlyTranslogManager(
+            translogManagerRef = new WriteOnlyInternalEngineTranslogManager(
                 engineConfig.getTranslogConfig(),
                 engineConfig.getPrimaryTermSupplier(),
                 engineConfig.getGlobalCheckpointSupplier(),

server/src/main/java/org/opensearch/index/translog/InternalTranslogManager.java → server/src/main/java/org/opensearch/index/translog/InternalEngineTranslogManager.java

@@ -35,7 +35,7 @@
  *
  * @opensearch.internal
  */
-public class InternalTranslogManager implements EngineTranslogManager {
+public class InternalEngineTranslogManager implements TranslogManager {
 
     private final ReleasableLock readLock;
     private final LifecycleAware engineLifeCycleAware;
@@ -46,7 +46,7 @@ public class InternalTranslogManager implements EngineTranslogManager {
     private final Supplier<LocalCheckpointTracker> localCheckpointTrackerSupplier;
     private final Logger logger;
 
-    public InternalTranslogManager(
+    public InternalEngineTranslogManager(
         TranslogConfig translogConfig,
         LongSupplier primaryTermSupplier,
         LongSupplier globalCheckpointSupplier,

server/src/main/java/org/opensearch/index/translog/NoOpInternalTranslogManager.java → server/src/main/java/org/opensearch/index/translog/NoOpEngineTranslogManager.java

@@ -19,14 +19,15 @@
 
 /**
  * A no-op {@link TranslogManager} implementation capable of interfacing with the {@link org.opensearch.index.engine.Engine}
+ * to be used when the engine does not require a translog.
  *
  * @opensearch.internal
  */
-public class NoOpInternalTranslogManager implements EngineTranslogManager {
+public class NoOpEngineTranslogManager implements TranslogManager {
     private final TranslogDeletionPolicy translogDeletionPolicy;
     private final String translogUUID;
 
-    public NoOpInternalTranslogManager(EngineConfig engineConfig, String translogUUID) {
+    public NoOpEngineTranslogManager(EngineConfig engineConfig, String translogUUID) {
         this.translogUUID = translogUUID;
         this.translogDeletionPolicy = new DefaultTranslogDeletionPolicy(
             engineConfig.getIndexSettings().getTranslogRetentionSize().getBytes(),

server/src/main/java/org/opensearch/index/translog/NoOpTranslogManager.java

@@ -11,6 +11,7 @@
 import org.opensearch.common.lease.Releasable;
 import org.opensearch.common.util.concurrent.ReleasableLock;
 import org.opensearch.core.index.shard.ShardId;
+import org.opensearch.index.seqno.SequenceNumbers;
 
 import java.io.IOException;
 import java.util.stream.Stream;
@@ -134,4 +135,40 @@ public Releasable drainSync() {
     public Translog.TranslogGeneration getTranslogGeneration() {
         return null;
     }
+
+    @Override
+    public long getLastSyncedGlobalCheckpoint() {
+        return 0;
+    }
+
+    @Override
+    public long getMaxSeqNo() {
+        return SequenceNumbers.NO_OPS_PERFORMED;
+    }
+
+    @Override
+    public void trimUnreferencedReaders() throws IOException {}
+
+    @Override
+    public boolean shouldPeriodicallyFlush(long localCheckpointOfLastCommit, long flushThreshold) {
+        return false;
+    }
+
+    @Override
+    public Exception getTragicExceptionIfClosed() {
+        return null;
+    }
+
+    @Override
+    public TranslogDeletionPolicy getDeletionPolicy() {
+        return null;
+    }
+
+    @Override
+    public String getTranslogUUID() {
+        return "";
+    }
+
+    @Override
+    public void close() throws IOException {}
 }

server/src/main/java/org/opensearch/index/translog/TranslogManager.java

@@ -11,6 +11,7 @@
 import org.opensearch.common.annotation.PublicApi;
 import org.opensearch.common.lease.Releasable;
 
+import java.io.Closeable;
 import java.io.IOException;
 import java.util.stream.Stream;
 
@@ -20,7 +21,7 @@
  * @opensearch.api
  */
 @PublicApi(since = "1.0.0")
-public interface TranslogManager {
+public interface TranslogManager extends Closeable {
 
     /**
      * Rolls the translog generation and cleans unneeded.
@@ -142,4 +143,46 @@ public interface TranslogManager {
     Releasable drainSync();
 
     Translog.TranslogGeneration getTranslogGeneration();
+
+    /**
+     * Retrieves last synced global checkpoint.
+     */
+    long getLastSyncedGlobalCheckpoint();
+
+    /**
+     * Retrieves the max seq no.
+     */
+    long getMaxSeqNo();
+
+    /**
+     * Trims unreferenced translog generations by asking {@link TranslogDeletionPolicy} for the minimum required
+     * generation.
+     */
+    void trimUnreferencedReaders() throws IOException;
+
+    /**
+     *
+     * @param localCheckpointOfLastCommit local checkpoint reference of last commit to translog
+     * @param flushThreshold threshold to flush to translog
+     * @return if the translog should be flushed
+     */
+    boolean shouldPeriodicallyFlush(long localCheckpointOfLastCommit, long flushThreshold);
+
+    /**
+     * Retrieves the underlying translog tragic exception
+     * @return the tragic exception
+     */
+    Exception getTragicExceptionIfClosed();
+
+    /**
+     * Retrieves the translog deletion policy
+     * @return TranslogDeletionPolicy
+     */
+    TranslogDeletionPolicy getDeletionPolicy();
+
+    /**
+     * Retrieves the translog unique identifier
+     * @return the uuid of the translog
+     */
+    String getTranslogUUID();
 }

server/src/main/java/org/opensearch/index/translog/WriteOnlyTranslogManager.java → server/src/main/java/org/opensearch/index/translog/WriteOnlyInternalEngineTranslogManager.java

@@ -24,9 +24,9 @@
  *
  * @opensearch.internal
  */
-public class WriteOnlyTranslogManager extends InternalTranslogManager {
+public class WriteOnlyInternalEngineTranslogManager extends InternalEngineTranslogManager {
 
-    public WriteOnlyTranslogManager(
+    public WriteOnlyInternalEngineTranslogManager(
         TranslogConfig translogConfig,
         LongSupplier primaryTermSupplier,
         LongSupplier globalCheckpointSupplier,