HDDS-8982. Infinite loop in WritableRatisContainerProvider if pipeline's nodes are not found

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ScmConfigKeys.java

@@ -614,6 +614,11 @@ public final class ScmConfigKeys {
       "ozone.scm.ha.dbtransactionbuffer.flush.interval";
   public static final long
       OZONE_SCM_HA_DBTRANSACTIONBUFFER_FLUSH_INTERVAL_DEFAULT = 600 * 1000L;
+
+  public static final String OZONE_SCM_GET_CONTAINER_MAX_RETRY =
+      "ozone.scm.get.container.max.retry";
+  public static final int OZONE_SCM_GET_CONTAINER_MAX_RETRY_DEFAULT = 4096;
+
   /**
    * Never constructed.
    */

hadoop-hdds/common/src/main/resources/ozone-default.xml

@@ -817,6 +817,14 @@
       ids will be allocated in a single batch.
     </description>
   </property>
+  <property>
+    <name>ozone.scm.get.container.max.retry</name>
+    <value>4096</value>
+    <tag>OZONE, SCM</tag>
+    <description>
+      This property determines how many times SCM is going to retry getContainer operation for Ratis type container.
+    </description>
+  </property>
   <property>
     <name>ozone.scm.chunk.size</name>
     <value>4MB</value>

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerManagerImpl.java

@@ -340,7 +340,10 @@ public ContainerInfo getMatchingContainer(final long size, final String owner,
       }
     } catch (Exception e) {
       LOG.warn("Container allocation failed on pipeline={}", pipeline, e);
-      return null;
+      return new ContainerInfo.Builder()
+          .setContainerID(-1)
+          .setPipelineID(pipeline.getId())
+          .build();
     }
   }
 

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/WritableECContainerProvider.java

@@ -207,6 +207,9 @@ private ContainerInfo allocateContainer(ReplicationConfig repConfig,
         excludedNodes, Collections.emptyList());
     ContainerInfo container =
         containerManager.getMatchingContainer(size, owner, newPipeline);
+    if (container != null && container.getContainerID() == -1) {
+      container = null;
+    }
     pipelineManager.openPipeline(newPipeline.getId());
     LOG.info("Created and opened new pipeline {}", newPipeline);
     return container;

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/WritableRatisContainerProvider.java

@@ -36,6 +36,9 @@
 import java.util.List;
 import java.util.stream.Collectors;
 
+import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_GET_CONTAINER_MAX_RETRY;
+import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_GET_CONTAINER_MAX_RETRY_DEFAULT;
+
 /**
  * Class to obtain a writable container for Ratis and Standalone pipelines.
  */
@@ -49,6 +52,8 @@ public class WritableRatisContainerProvider
   private final PipelineManager pipelineManager;
   private final PipelineChoosePolicy pipelineChoosePolicy;
   private final ContainerManager containerManager;
+  private int maxRetryGetContainer;
+
 
   public WritableRatisContainerProvider(ConfigurationSource conf,
       PipelineManager pipelineManager,
@@ -58,6 +63,8 @@ public WritableRatisContainerProvider(ConfigurationSource conf,
     this.pipelineManager = pipelineManager;
     this.containerManager = containerManager;
     this.pipelineChoosePolicy = pipelineChoosePolicy;
+    this.maxRetryGetContainer = conf.getInt(OZONE_SCM_GET_CONTAINER_MAX_RETRY,
+        OZONE_SCM_GET_CONTAINER_MAX_RETRY_DEFAULT);
   }
 
 
@@ -85,8 +92,8 @@ public ContainerInfo getContainer(final long size,
     //TODO we need to continue the refactor to use repConfig everywhere
     //in downstream managers.
 
-
-    while (true) {
+    int currentCount = 0;
+    while (currentCount < maxRetryGetContainer) {
       List<Pipeline> availablePipelines;
       Pipeline pipeline;
       // Acquire pipeline manager lock, to avoid any updates to pipeline
@@ -103,7 +110,13 @@ public ContainerInfo getContainer(final long size,
               excludeList);
         }
         if (containerInfo != null) {
-          return containerInfo;
+          // if containerID == -1, means Container allocation
+          // failed on selected pipeline
+          if (containerInfo.getContainerID() != -1) {
+            return containerInfo;
+          } else {
+            excludeList.addPipeline(containerInfo.getPipelineID());
+          }
         }
       } finally {
         pipelineManager.releaseReadLock();
@@ -164,12 +177,17 @@ public ContainerInfo getContainer(final long size,
           containerInfo = selectContainer(availablePipelines, size, owner,
               excludeList);
           if (containerInfo != null) {
-            return containerInfo;
+            if (containerInfo.getContainerID() != -1) {
+              return containerInfo;
+            } else {
+              excludeList.addPipeline(containerInfo.getPipelineID());
+            }
           }
         } finally {
           pipelineManager.releaseReadLock();
         }
       }
+      currentCount++;
     }
 
     // we have tried all strategies we know but somehow we are not able
@@ -197,7 +215,7 @@ private List<Pipeline> findPipelinesByState(
     return pipelines;
   }
 
-  private ContainerInfo selectContainer(List<Pipeline> availablePipelines,
+  protected ContainerInfo selectContainer(List<Pipeline> availablePipelines,
       long size, String owner, ExcludeList excludeList) {
     Pipeline pipeline;
     ContainerInfo containerInfo;

hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestWritableRatisContainerProvider.java

@@ -0,0 +1,182 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with this
+ * work for additional information regarding copyright ownership.  The ASF
+ * licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.hadoop.hdds.scm.pipeline;
+
+import org.apache.hadoop.hdds.client.ReplicationConfig;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.scm.PipelineChoosePolicy;
+import org.apache.hadoop.hdds.scm.PipelineRequestInformation;
+import org.apache.hadoop.hdds.scm.container.ContainerInfo;
+import org.apache.hadoop.hdds.scm.container.ContainerManager;
+import org.apache.hadoop.hdds.scm.container.common.helpers.ExcludeList;
+import org.apache.hadoop.hdds.scm.pipeline.choose.algorithms.HealthyPipelineChoosePolicy;
+import org.apache.hadoop.hdds.scm.pipeline.choose.algorithms.RandomPipelineChoosePolicy;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+import org.junit.runners.Parameterized;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.UUID;
+
+import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_GET_CONTAINER_MAX_RETRY;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyCollection;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anySet;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+
+/**
+ * Tests to validate the WritableRatisContainerProvider works correctly.
+ */
+public class TestWritableRatisContainerProvider {
+
+  private static final String OWNER = "SCM";
+  private PipelineManager pipelineManager;
+  private ContainerManager containerManager;
+  private OzoneConfiguration conf;
+  private WritableRatisContainerProvider provider;
+  private ReplicationConfig repConfig;
+
+  private int testSCMGetContainerMaxRetry;
+
+  @Parameterized.Parameters
+  public static Collection<PipelineChoosePolicy> policies() {
+    Collection<PipelineChoosePolicy> policies = new ArrayList<>();
+
+    Pipeline pipeline = mock(Pipeline.class);
+    RandomPipelineChoosePolicy randomPipelineChoosePolicy =
+        mock(RandomPipelineChoosePolicy.class);
+    when(randomPipelineChoosePolicy.choosePipeline(anyList(),
+        any(PipelineRequestInformation.class))).thenReturn(pipeline);
+
+    HealthyPipelineChoosePolicy healthyPipelineChoosePolicy =
+        mock(HealthyPipelineChoosePolicy.class);
+    when(healthyPipelineChoosePolicy.choosePipeline(anyList(),
+        any(PipelineRequestInformation.class))).thenReturn(pipeline);
+
+    policies.add(randomPipelineChoosePolicy);
+    policies.add(healthyPipelineChoosePolicy);
+
+    return policies;
+  }
+
+  void setup(PipelineChoosePolicy policy) {
+    conf = new OzoneConfiguration();
+    testSCMGetContainerMaxRetry = 3;
+    conf.setInt(OZONE_SCM_GET_CONTAINER_MAX_RETRY,
+        testSCMGetContainerMaxRetry);
+
+    repConfig = mock(ReplicationConfig.class);
+    pipelineManager = mock(PipelineManager.class);
+    containerManager = mock(ContainerManager.class);
+    provider = spy(new WritableRatisContainerProvider(conf,
+        pipelineManager, containerManager, policy));
+  }
+
+  @ParameterizedTest
+  @MethodSource("policies")
+  public void testSelectContainerShouldNoMoreThanMaxRetry(
+      PipelineChoosePolicy policy) {
+    setup(policy);
+    List<Pipeline> pipelines = mock(ArrayList.class);
+    when(pipelines.size()).thenReturn(3);
+    when(pipelineManager.getPipelines(any(ReplicationConfig.class),
+        any(Pipeline.PipelineState.class), anyCollection(),
+        anyCollection())).thenReturn(pipelines);
+    ContainerInfo containerInfo = mock(ContainerInfo.class);
+    when(containerInfo.getContainerID()).thenReturn(-1L);
+    when(containerManager.getMatchingContainer(anyLong(), anyString(),
+        any(Pipeline.class), anySet())).thenReturn(containerInfo);
+
+    ExcludeList exclude = mock(ExcludeList.class);
+
+    try {
+      provider.getContainer(
+          1, repConfig, OWNER, exclude);
+      fail("expected IOException");
+    } catch (IOException e) {
+      assertTrue(e.getMessage()
+          .contains("Unable to allocate a container to the block"));
+    }
+
+    verify(provider, times(testSCMGetContainerMaxRetry))
+        .selectContainer(anyList(), anyLong(),
+            anyString(), any(ExcludeList.class));
+  }
+
+
+  @ParameterizedTest
+  @MethodSource("policies")
+  public void testSelectContainerShouldNoMoreThanMaxRetryAfterCreateNewPipeline(
+      PipelineChoosePolicy policy) throws IOException {
+    setup(policy);
+    List<Pipeline> emptyPipelines = new ArrayList<>();
+    List<Pipeline> pipelines = mock(ArrayList.class);
+    when(pipelines.size()).thenReturn(3);
+    when(pipelineManager.getPipelines(any(ReplicationConfig.class),
+        any(Pipeline.PipelineState.class), anyCollection(), anyCollection()))
+        .thenReturn(emptyPipelines, pipelines);
+
+    Pipeline pipeline1 = mock(Pipeline.class);
+    when(pipeline1.getId()).thenReturn(mock(PipelineID.class));
+    when(pipelineManager.createPipeline(any(ReplicationConfig.class)))
+        .thenReturn(pipeline1);
+
+    ContainerInfo containerInfo = mock(ContainerInfo.class);
+    when(containerInfo.getContainerID()).thenReturn(-1L);
+
+    PipelineID pipelineID2 = PipelineID.valueOf(UUID.randomUUID());
+
+    when(containerInfo.getPipelineID()).thenReturn(pipelineID2);
+    when(containerManager.getMatchingContainer(anyLong(), anyString(), any(
+        Pipeline.class), anySet())).thenReturn(containerInfo);
+
+    ExcludeList exclude = new ExcludeList();
+
+    try {
+      provider.getContainer(
+          1, repConfig, OWNER, exclude);
+      fail("expected IOException");
+    } catch (IOException e) {
+      assertTrue(e.getMessage().
+          contains("Unable to allocate a container to the block"));
+    }
+
+    assertEquals(exclude.getPipelineIds(),
+        new HashSet<>(Arrays.asList(pipelineID2)));
+    verify(provider, times(testSCMGetContainerMaxRetry))
+        .selectContainer(anyList(), anyLong(), anyString(),
+            any(ExcludeList.class));
+  }
+
+}