static DatanodeStorageInfo[] chooseTargetForNewBlock(
      BlockManager bm, String src, DatanodeInfo[] excludedNodes,
      String[] favoredNodes, EnumSet<AddBlockFlag> flags,
      ValidateAddBlockResult r) throws IOException {
    Node clientNode = null;

    boolean ignoreClientLocality = (flags != null
            && flags.contains(AddBlockFlag.IGNORE_CLIENT_LOCALITY));

    // If client locality is ignored, clientNode remains 'null' to indicate
    if (!ignoreClientLocality) {
      clientNode = bm.getDatanodeManager().getDatanodeByHost(r.clientMachine);
      if (clientNode == null) {
        clientNode = getClientNode(bm, r.clientMachine);
      }
    }

    Set<Node> excludedNodesSet =
        (excludedNodes == null) ? new HashSet<>()
            : new HashSet<>(Arrays.asList(excludedNodes));

    List<String> favoredNodesList =
        (favoredNodes == null) ? Collections.emptyList()
            : Arrays.asList(favoredNodes);

    // choose targets for the new block to be allocated.
    return bm.chooseTarget4NewBlock(src, r.numTargets, clientNode,
                                    excludedNodesSet, r.blockSize,
                                    favoredNodesList, r.storagePolicyID,
                                    r.blockType, r.ecPolicy, flags);
  }
@Test
  @SuppressWarnings("unchecked")
  public void testIgnoreClientLocality() throws IOException {
    ValidateAddBlockResult addBlockResult =
        new ValidateAddBlockResult(1024L, 3, (byte) 0x01, null, null, null);

    EnumSet<AddBlockFlag> addBlockFlags =
        EnumSet.of(AddBlockFlag.IGNORE_CLIENT_LOCALITY);

    BlockManager bmMock = mock(BlockManager.class);

    ArgumentCaptor<Node> nodeCaptor = ArgumentCaptor.forClass(Node.class);

    when(bmMock.chooseTarget4NewBlock(anyString(), anyInt(), any(), anySet(),
        anyLong(), anyList(), anyByte(), any(), any(), any())).thenReturn(null);

    FSDirWriteFileOp.chooseTargetForNewBlock(bmMock, "localhost", null, null,
        addBlockFlags, addBlockResult);

    // There should be no other interactions with the block manager when the
    // IGNORE_CLIENT_LOCALITY is passed in because there is no need to discover
    // the local node requesting the new block
    verify(bmMock, times(1)).chooseTarget4NewBlock(anyString(), anyInt(),
        nodeCaptor.capture(), anySet(), anyLong(), anyList(), anyByte(), any(),
        any(), any());

    verifyNoMoreInteractions(bmMock);

    assertNull(
        "Source node was assigned a value. Expected 'null' value because "
            + "chooseTarget was flagged to ignore source node locality",
        nodeCaptor.getValue());
  }