recon

Author: guohao-rosicky

hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/scm/ReconPipelineManager.java

@@ -21,6 +21,7 @@
 import java.io.IOException;
 import java.time.Clock;
 import java.time.ZoneOffset;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.stream.Collectors;
 
@@ -99,35 +100,49 @@ public static ReconPipelineManager newReconPipelineManager(
    */
   void initializePipelines(List<Pipeline> pipelinesFromScm)
       throws IOException {
-
-
-    List<Pipeline> pipelinesInHouse = getPipelines();
-    LOG.info("Recon has {} pipelines in house.", pipelinesInHouse.size());
-    for (Pipeline pipeline : pipelinesFromScm) {
-
-      ReplicationConfig replicationConfig = pipeline.getReplicationConfig();
-      acquireWriteLock(replicationConfig);
-      try {
-        // New pipeline got from SCM. Validate If it doesn't exist at Recon, try adding it.
-        if (addPipeline(pipeline)) {
-          LOG.info("Added new pipeline {} from SCM.", pipeline.getId());
-        } else {
-          LOG.warn("Pipeline {} already exists in Recon pipeline metadata.",
-              pipeline.getId());
-          // Recon already has this pipeline. Just update state and creation
-          // time.
-          getStateManager().updatePipelineState(pipeline.getId().getProtobuf(),
-              Pipeline.PipelineState.getProtobuf(pipeline.getPipelineState()));
-          getPipeline(pipeline.getId()).setCreationTimestamp(
-              pipeline.getCreationTimestamp());
+    List<HddsProtos.ReplicationType> replicationTypes = new ArrayList<>();
+    replicationTypes.add(HddsProtos.ReplicationType.EC);
+    replicationTypes.add(HddsProtos.ReplicationType.RATIS);
+
+    for (HddsProtos.ReplicationType replicationType : replicationTypes) {
+      List<Pipeline> pipelines = pipelinesFromScm.stream().filter(
+          p -> p.getReplicationConfig().getReplicationType()
+              .equals(replicationType)).collect(Collectors.toList());
+
+      if (!pipelines.isEmpty()) {
+        ReplicationConfig replicationConfig =
+            pipelines.iterator().next().getReplicationConfig();
+
+        acquireWriteLock(replicationConfig);
+        try {
+          List<Pipeline> pipelinesInHouse = getPipelines().stream().filter(
+              p -> p.getReplicationConfig().getReplicationType()
+                  .equals(replicationType)).collect(Collectors.toList());
+
+          LOG.info("Recon has {} pipelines in house.", pipelinesInHouse.size());
+          for (Pipeline pipeline : pipelines) {
+            // New pipeline got from SCM. Validate If it doesn't exist at Recon, try adding it.
+            if (addPipeline(pipeline)) {
+              LOG.info("Added new pipeline {} from SCM.", pipeline.getId());
+            } else {
+              LOG.warn("Pipeline {} already exists in Recon pipeline metadata.",
+                  pipeline.getId());
+              // Recon already has this pipeline. Just update state and creation
+              // time.
+              getStateManager().updatePipelineState(
+                  pipeline.getId().getProtobuf(),
+                  Pipeline.PipelineState.getProtobuf(
+                      pipeline.getPipelineState()));
+              getPipeline(pipeline.getId()).setCreationTimestamp(
+                  pipeline.getCreationTimestamp());
+            }
+          }
+          removeInvalidPipelines(pipelines);
+        } finally {
+          releaseWriteLock(replicationConfig);
         }
-      } finally {
-        releaseWriteLock(replicationConfig);
       }
-
     }
-    removeInvalidPipelines(pipelinesFromScm);
-
   }
 
   public void removeInvalidPipelines(List<Pipeline> pipelinesFromScm) {
@@ -141,20 +156,22 @@ public void removeInvalidPipelines(List<Pipeline> pipelinesFromScm) {
         .collect(Collectors.toList());
     invalidPipelines.forEach(p -> {
       PipelineID pipelineID = p.getId();
-      ReplicationConfig replicationConfig = p.getReplicationConfig();
-      acquireWriteLock(replicationConfig);
-      try {
-        if (!p.getPipelineState().equals(CLOSED)) {
-          getStateManager().updatePipelineState(pipelineID.getProtobuf(),
+      if (!p.getPipelineState().equals(CLOSED)) {
+        try {
+          getStateManager().updatePipelineState(
+              pipelineID.getProtobuf(),
               HddsProtos.PipelineState.PIPELINE_CLOSED);
+        } catch (IOException e) {
+          LOG.warn("Pipeline {} not found while updating state. ",
+              p.getId(), e);
         }
+      }
+      try {
         LOG.info("Removing invalid pipeline {} from Recon.", pipelineID);
         closePipeline(p.getId());
         deletePipeline(p.getId());
       } catch (IOException e) {
         LOG.warn("Unable to remove pipeline {}", pipelineID, e);
-      } finally {
-        releaseWriteLock(replicationConfig);
       }
     });
   }