refactoring

Signed-off-by: Poojita Raj <[email protected]>
opensearch-project · May 31, 2023 · 1727637 · 1727637
1 parent 21f7a75
commit 1727637
Show file tree

Hide file tree

Showing 12 changed files with 119 additions and 131 deletions.
diff --git a/server/src/main/java/org/opensearch/indices/replication/SegmentReplicationSourceService.java b/server/src/main/java/org/opensearch/indices/replication/SegmentReplicationSourceService.java
@@ -11,11 +11,13 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.message.ParameterizedMessage;
+import org.apache.lucene.store.AlreadyClosedException;
 import org.opensearch.action.support.ChannelActionListener;
 import org.opensearch.cluster.ClusterChangedEvent;
 import org.opensearch.cluster.ClusterStateListener;
 import org.opensearch.cluster.metadata.IndexMetadata;
 import org.opensearch.cluster.node.DiscoveryNode;
+import org.opensearch.cluster.node.DiscoveryNodes;
 import org.opensearch.cluster.routing.ShardRouting;
 import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.common.Nullable;
@@ -37,7 +39,9 @@
 import org.opensearch.transport.TransportService;
 
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicLong;
 
@@ -170,6 +174,35 @@ public void clusterChanged(ClusterChangedEvent event) {
                 }
             }
         }
+        if (event.nodesChanged()) {
+            List<IndexShard> indexShardList = new ArrayList<>();
+            DiscoveryNodes nodes = event.state().nodes();
+            if (nodes.getMinNodeVersion().equals(nodes.getMaxNodeVersion())) {
+                for (IndexService indexService : indicesService) {
+                    if (indexService.getIndexSettings().isSegRepEnabled() && (indexService.getIndexSettings().getNumberOfReplicas() > 0)) {
+                        for (IndexShard indexShard : indexService) {
+                            try {
+                                if (indexShard.routingEntry().primary()
+                                    && (indexShard.getEngine().config().getClusterMinVersion() != nodes.getMaxNodeVersion())) {
+                                    indexShardList.add(indexShard);
+                                }
+                            } catch (AlreadyClosedException e) {
+                                logger.warn("Index shard [{}] engine is already closed.", indexShard.shardId());
+                            }
+                        }
+                    }
+                }
+            }
+            try {
+                if (indexShardList.isEmpty() == false) {
+                    for (IndexShard indexShard : indexShardList) {
+                        indexShard.resetEngine();
+                    }
+                }
+            } catch (Exception e) {
+                logger.error("Received unexpected exception: [{}]", e.getMessage());
+            }
+        }
     }
 
     @Override

diff --git a/...r/src/main/java/org/opensearch/indices/replication/SegmentReplicationUpgradeListener.java b/...r/src/main/java/org/opensearch/indices/replication/SegmentReplicationUpgradeListener.java
diff --git a/...er/src/main/java/org/opensearch/indices/replication/SegmentReplicationUpgradeService.java b/...er/src/main/java/org/opensearch/indices/replication/SegmentReplicationUpgradeService.java
diff --git a/server/src/main/java/org/opensearch/node/Node.java b/server/src/main/java/org/opensearch/node/Node.java
@@ -52,7 +52,6 @@
 import org.opensearch.indices.replication.SegmentReplicationSourceService;
 import org.opensearch.extensions.ExtensionsManager;
 import org.opensearch.extensions.NoopExtensionsManager;
-import org.opensearch.indices.replication.SegmentReplicationUpgradeService;
 import org.opensearch.monitor.fs.FsInfo;
 import org.opensearch.monitor.fs.FsProbe;
 import org.opensearch.plugins.ExtensionAwarePlugin;
@@ -1098,8 +1097,6 @@ protected Node(
                     b.bind(SegmentReplicationSourceService.class)
                         .toInstance(new SegmentReplicationSourceService(indicesService, transportService, recoverySettings));
                 }
-                b.bind(SegmentReplicationUpgradeService.class)
-                    .toInstance(new SegmentReplicationUpgradeService(indicesService, clusterService.getClusterApplierService()));
                 b.bind(HttpServerTransport.class).toInstance(httpServerTransport);
                 pluginComponents.stream().forEach(p -> b.bind((Class) p.getClass()).toInstance(p));
                 b.bind(PersistentTasksService.class).toInstance(persistentTasksService);