#!/bin/bash

# Copyright (c) 2011-2012 Cloudera, Inc. All rights reserved.

# Time marker for both stderr and stdout
date; date 1>&2

cloudera_config=`dirname $0`
cloudera_config=`cd "$cloudera_config/../common"; pwd`
. ${cloudera_config}/cloudera-config.sh

# load the parcel environment
source_parcel_environment

# attempt to find java
locate_cdh_java_home

# simulate /etc/default/hadoop if necessary
. ${cloudera_config}/cdh-default-hadoop

# replace {{PID}} in HeapDumpPath option
export HADOOP_NAMENODE_OPTS=$(replace_pid $HADOOP_NAMENODE_OPTS)
export HADOOP_DATANODE_OPTS=$(replace_pid $HADOOP_DATANODE_OPTS)
export HADOOP_SECONDARYNAMENODE_OPTS=$(replace_pid $HADOOP_SECONDARYNAMENODE_OPTS)
export HADOOP_NFS3_OPTS=$(replace_pid $HADOOP_NFS3_OPTS)
export HADOOP_JOURNALNODE_OPTS=$(replace_pid $HADOOP_JOURNALNODE_OPTS)

if [ "$CDH_VERSION" -ge "4" ]; then
  HDFS_BIN=$HADOOP_HDFS_HOME/bin/hdfs

  # Disable IPv6. This is automatically done by bin/hadoop for CDH3.
  export HADOOP_OPTS="-Djava.net.preferIPv4Stack=true $HADOOP_OPTS"
elif [ "3" = "$CDH_VERSION" ] || [ "-3" = "$CDH_VERSION" ]; then
  HDFS_BIN=$CDH_HADOOP_HOME/bin/hadoop
  export HADOOP_HOME=$CDH_HADOOP_HOME
  echo "using $HADOOP_HOME as HADOOP_HOME"
else
  echo "ERROR: Unsupported version $CDH_VERSION"
  exit 1
fi

echo "using $JAVA_HOME as JAVA_HOME"
echo "using $CDH_VERSION as CDH_VERSION"
echo "using $CONF_DIR as CONF_DIR"
echo "using $SECURE_USER as SECURE_USER"
echo "using $SECURE_GROUP as SECURE_GROUP"

set_hadoop_classpath

# debug
set -x

function connect_to_port() {
  PORT=$1
  python -c 'import socket; import sys; s = socket.socket(socket.AF_INET); s.settimeout(5.0); s.connect((sys.argv[1], int(sys.argv[2]))); s.close();' localhost $PORT
  if [ $? -ne 0 ]; then
    echo "Cannot connect to port $PORT."
    return 1
  fi
  return 0
}

function set_jsvc_home() {
  # We have decided to put jsvc in different locations on Centos and Sles.
  # So we need to check what we set by default and at the alternative location
  # too to make sure we have the right one.
  if [[ ! -e $JSVC_HOME/jsvc ]]; then
    export JSVC_HOME=/usr/lib/bigtop-utils
    if [[ ! -e $JSVC_HOME/jsvc ]]; then
      echo "No jsvc file found."
      exit 1
    fi
  fi
  echo "using $JSVC_HOME as JSVC_HOME"
}

# Search-replace {{CMF_CONF_DIR}} in files
replace_conf_dir

# Make sure topology.py and cloudera_manager_agent_fencer.py (where applicable) are executable
make_scripts_executable

# Set max locked memory for DataNode caching
if [ "DATANODE_MAX_LOCKED_MEMORY" != "" ]; then
  ulimit -l $DATANODE_MAX_LOCKED_MEMORY
fi
# if HADOOP_IDENT_STRING is not set, $USER will be used instead. There have
# been situations where $USER is not set for su which results in
# hadoop.id.str ending up to be an empty string so we're setting it here
# explicitly
export HADOOP_IDENT_STRING="hdfs"

# Secure datanode requires additonal env vars.
if [ -n "$SECURE_DATANODE" ]; then
  if [ "$CDH_VERSION" -ge "4" ]; then
    export HADOOP_SECURE_DN_USER=$SECURE_USER
    echo "using $HADOOP_SECURE_DN_USER as HADOOP_SECURE_DN_USER"
    set_jsvc_home
  elif [ "3" = "$CDH_VERSION" ] || [ "-3" = "$CDH_VERSION" ]; then
    export HADOOP_DATANODE_USER=$SECURE_USER
    echo "using $HADOOP_DATANODE_USER as HADOOP_DATANODE_USER"
  fi
  # We're running as a daemon. The agent spawns us as root:root but HDFS_BIN
  # will impersonate another user, so file ownership must be reset. 
  chown -R $SECURE_USER:$SECURE_GROUP $CONF_DIR
fi

# If doing fast mkdir, kerberos login is handled by
# the CdhClient.
if [ "mkdir" != "$1" ] || [ -z "$FAST_MKDIR" ]; then
  acquire_kerberos_tgt hdfs.keytab
fi


if [ "validate-writable-empty-dirs" = "$1" ]; then
  if [ -z "$VALIDATE_DIRS" ]; then
    echo "No namenode dirs specified." # unexpected state; CMF bug.
    exit 1
  fi
  for validate_dir in $VALIDATE_DIRS; do
    if [ -e $validate_dir ]; then
      if [[ ! -d $validate_dir || ! -w $validate_dir ]]; then
        echo "Validation failed: $validate_dir must be a directory and writable by hdfs."
        exit 1
      fi
      NON_EMPTY_DIR=`find -L "$validate_dir" -maxdepth 0 -not -empty`
      if [ -n "$NON_EMPTY_DIR" ]; then
        if [ "$2" = "true" ]; then
          echo "$validate_dir is not empty, clearing it"
          rm -rf $validate_dir/*
        else
          echo "Validation failed: $validate_dir exists, but is not empty"
          exit 1
        fi
      fi
    fi
  done
  echo "Validation of $VALIDATE_DIRS succeeded."
  exit 0
elif [ "file-operation" = "$1" ]; then
  TARGET_DIR=$2
  FILENAME=$3
  COMMAND=$4
  FULL_PATH=$TARGET_DIR/$FILENAME
  if [ "WRITE" = "$COMMAND" ]; then
    cat > $FULL_PATH <<< $FILENAME
    RET=$?
    if [ $RET -eq 0 ]; then
      echo "Successfully wrote $FULL_PATH"
    else
      echo "Failed to write $FULL_PATH"
    fi
 elif [ "CONFIRM_EXISTS" = "$COMMAND" ]; then
   # Confirm existence of a file by trying to write the same file
   # when clobbering is not allowed. Writing is preferred over reading
   # for NFS, where sync on write should be configured.
   set -o noclobber
   cat > $FULL_PATH <<< $FILENAME
   RET=$?
   if [ $RET -eq 1 ]; then
     # Return code of 1 when cannot overwrite existing file
     echo "Successfully confirmed that file already exists: $FULL_PATH"
     RET=0
   elif [ $RET -eq 0 ]; then
     echo "File did not exist: $FULL_PATH. Make sure that $TARGET_DIR is shared and is configured to sync on write."
     rm $FULL_PATH
     RET=1
   else
     echo "Failed to confirm whether file exists: $FULL_PATH"
   fi
   set +o noclobber
 elif [ "DELETE" = "$COMMAND" ]; then
    rm $FULL_PATH
    RET=$?
    if [ $RET -eq 0 ]; then
      echo "Successfully deleted $FULL_PATH"
    else
      echo "Failed to delete $FULL_PATH"
    fi
  else
    echo "Illegal command: $COMMAND"
    RET=1
  fi
  exit $RET
elif [ "bootstrap" = "$1" ]; then
  FORCE_ARG=""
  if [ "force" = "$2" ]; then
    FORCE_ARG="-force"
  fi
  exec $HDFS_BIN --config $CONF_DIR namenode -bootstrapStandby -nonInteractive $FORCE_ARG
elif [ "failover" = "$1" ]; then
  ACTIVE="Failover failed: Can't failover to an active service"
  NS=$2
  FROM_NN=$3
  TO_NN=$4
  FORCE=$5
  if [ "$FORCE" = "true" ]; then
    FORCE_FAILOVER="--forceactive"
  fi
  OUTPUT=$( $HDFS_BIN --config $CONF_DIR haadmin -ns $NS -failover $FORCE_FAILOVER $FROM_NN $TO_NN 2>&1 )
  RET=$?
  if [ $RET -eq 0 ]; then
    echo "Failed over from $FROM_NN to $TO_NN of nameservice $NS"
    exit 0
  elif ( grep -q "$ACTIVE" <<< "$OUTPUT" ); then
    echo "Failover from $FROM_NN to $TO_NN of nameservice $NS unnecessary as $TO_NN is already active."
    exit 2
  else
    echo "Unable to failover from $FROM_NN to $TO_NN of nameservice $NS; see stderr log."
    if [ $RET -eq 2 ]; then
      # Return code of 2 is already used to indicate an acceptable failure.
      exit 1
    else
      exit $RET
    fi
  fi
elif [ "transition-to-active" = $1 ]; then
  if [ -z "$2" || -z "$3" ]; then
    echo "Nameservice and NameNode ID must be specified." # unexpected state; CM bug.
    exit 1
  fi
  NS=$2
  NN=$3
  exec $HDFS_BIN --config $CONF_DIR haadmin -ns $NS -transitionToActive $NN
elif [ "initializeSharedEdits" = "$1" ]; then
  exec $HDFS_BIN --config $CONF_DIR namenode -initializeSharedEdits -nonInteractive
elif [ "initialize-znode" = "$1" ]; then
  # $2 is either -force or empty
  exec $HDFS_BIN --config $CONF_DIR zkfc -formatZK -nonInteractive $2
elif [ "format-namenode" = "$1" ]; then
  if [ -z "$DFS_STORAGE_DIRS" ]; then
    echo "No storage dirs specified." # unexpected state; CM bug.
    exit 1
  fi

  # Validate all configured dirs are actually dirs.
  for dfsdir in $DFS_STORAGE_DIRS; do
    if [ -e $dfsdir ]; then
      if [ ! -d $dfsdir ]; then
        echo "Refusing to format.  $dfsdir is not a directory."
        exit 1
      fi
    fi
  done

  CLUSTER_ARGS=""
  if [ $# -eq 2 ]; then
    CLUSTER_ARGS="-clusterId $2"
  fi

  if [ "3" = "$CDH_VERSION" ] || [ "-3" = "$CDH_VERSION" ]; then
    # hadoop namenode -format helpfully double-checks your
    # request if the storage directories already exist.
    # In CM, they often do exist but are empty (because
    # we've pre-created them).
    # So, we force the format, as long as all the directories
    # are empty.
    for dfsdir in $DFS_STORAGE_DIRS; do
      if [ -e $dfsdir ]; then
        NON_EMPTY_DIR=`find -L "$dfsdir" -maxdepth 0 -not -empty`
        if [ -n "$NON_EMPTY_DIR" ]; then
          echo "Refusing to format: $dfsdir exists and is non-empty"
          cm_error_and_exit HDFS_DFS_DIR_NOT_EMPTY
        fi
      fi
    done

    /usr/bin/yes Y 2> /dev/null | exec $HDFS_BIN --config $CONF_DIR namenode -format $CLUSTER_ARGS
  else

    # Since CDH4.0.0 we can use -nonInteractive instead of the yes pipe.
    # This mode will also make sure the dirs are empty, including any JN dirs.
    exec $HDFS_BIN --config $CONF_DIR namenode -format $CLUSTER_ARGS -nonInteractive
  fi

elif [ "monitor-decommission" = "$1" ]; then
  # This cannot be used with CDH < 4 since report format has changed.
  if [ $CDH_VERSION -lt 4 ]; then
    echo "Monitor decommission cannot be used with CDH 3"
    exit 1
  fi
  # $2 is the list of names to look for in a '|' delimited list.
  # $3 is the exact NameNode to monitor, in the format of hdfs://host:port
  # Extract hostnames of DNs being decommissioned
  WAIT=0
  while [ $WAIT -eq 0 ]; do
    # There's nothing special about 15 seconds. It's just attempting to
    # strike a balance between too much polling and avoiding a large
    # artificial delay in the case that the operation is quick.
    sleep 15
    # Yay for robust interfaces.
    # Each host gets a report block in the output. Line 2 is the Hostname and
    # line 4 is the decommissioning state. So, we find in progress hosts,
    # then extract the hostname.
    HOSTS=`$HDFS_BIN --config $CONF_DIR dfsadmin -fs $3 -report | grep 'Decommission in progress' -B2 | grep 'Hostname:' | cut -d ' ' -f 2`
    echo $HOSTS | grep -q -E "$2"
    WAIT=$?
  done
elif [ "jnSyncWait" = "$1" ]; then
  # Wait for all the journal nodes to get in sync with the edits of this namenode.
  # $2 is the NN's /jmx URL which is parsed to find out how many JNs are in sync.
  # $3 is the number of JN's in HDFS. We wait till this number of JNs are in sync.
  SYNC_COUNT=0
  ADD_IMPORTS="from cmf.url_util import urlopen_with_timeout;"
  while [ "$SYNC_COUNT" != "$3" ]; do
    sleep 15
    SYNC_COUNT=`python -c "$ADD_IMPORTS print urlopen_with_timeout('$2').read()" | grep IsOutOfSync | grep -c false`
  done
elif [ "nnRpcWait" = "$1" ]; then
  # Wait for namenode to start responding to RPCs, by checking that we can
  # get safemode status successfully.
  WAIT=1
  while [ $WAIT -ne 0 ]; do
    sleep 15
    $HDFS_BIN --config $CONF_DIR dfsadmin -fs "$2" -safemode get
    WAIT=$?
  done
elif [ "-safemode" = "$4" -a "get" = "$5" ]; then
  # Wait for NN to get out of safemode.
  # Though NN has a command to wait for safemode to be off, it's not very robust.
  # So we are going to poll for safemode status and wait till it is off.
  WAIT=1
  # We allow 4 failures in getting safemode status to handle transient network issues.
  FAIL_COUNT=0
  while [ $WAIT -ne 0 ]; do
    sleep 15
    STATUS=`$HDFS_BIN --config $CONF_DIR "$@"`
    if [ $? -ne 0 ]; then
      echo "Could not get safemode status"
      FAIL_COUNT=$(($FAIL_COUNT+1))
      if [ $FAIL_COUNT -ge 4 ]; then
        echo "Failed to get safemode status 4 times in a row. Exiting ... "
        exit 1
      fi
    else
      FAIL_COUNT=0 # Reset fail count
    fi
    # Wait till safemode is off
    echo $STATUS | grep OFF
    WAIT=$?
  done
elif [ "monitor-upgrade" = "$1" ]; then
  RET=0
  WAIT=0
  while [ $WAIT -eq 0 ]; do
    # See above
    sleep 15
    REPORT=`$HDFS_BIN --config $CONF_DIR dfsadmin -upgradeProgress status`
    # XXX: Check $? here and do something?
    echo $REPORT | grep -q -E "Upgrade for version .* has been completed"
    WAIT=$?
    RET=$WAIT
  done
  exit $RET
elif [ "finalize-upgrade" = "$1" ]; then
  exec $HDFS_BIN --config $CONF_DIR dfsadmin -finalizeUpgrade
elif [ "rolling-upgrade-prepare" = "$1" ]; then
  $HDFS_BIN --config $CONF_DIR dfsadmin -rollingUpgrade prepare
  RET=$?
  if [ $RET -ne 0 ]; then
    exit $RET
  fi
  WAIT=1
  while [ $WAIT -ne 0 ]; do
    sleep 15
    QUERY=`$HDFS_BIN --config $CONF_DIR dfsadmin -rollingUpgrade query`
    RET=$?
    if [ $RET -ne 0 ]; then
      exit $RET
    fi
    echo $QUERY | grep -q "Proceed with rolling upgrade"
    WAIT=$?
  done
  exit 0
elif [ "rolling-upgrade-finalize" = "$1" ]; then
  exec $HDFS_BIN --config $CONF_DIR dfsadmin -rollingUpgrade finalize
elif [ "nnDnLiveWait" = "$1" ]; then
  # $2 is the NameNode's /jmx URL
  # $3 and following are hostnames of DataNodes to check for liveness
  if [ $# -lt 3 ]; then
    echo "No DataNode hostnames specified."
    exit 1
  fi
  NN_JMX="$2"

  # This initial sleep is to avoid a race where a DN has been restarted
  # so recently that the NN reports contact with it within a 30-second
  # window (see below), but that contact is from before the restart.
  # We sleep 30 seconds here + 15 at the top of the loop, so we know
  # a last contact of 30 seconds ago or less occurred well after the
  # DN restart.
  sleep 30

  ADD_IMPORTS="from cmf.url_util import urlopen_with_timeout; import simplejson; "
  GET_LAST_CONTACT_BASE="nnInfo = simplejson.loads(urlopen_with_timeout('$NN_JMX').read());"
  GET_LAST_CONTACT_BASE="$GET_LAST_CONTACT_BASE liveNodes = simplejson.loads(nnInfo['beans'][0]['LiveNodes']); "

  ALL_DNS_LIVE=0
  while [ $ALL_DNS_LIVE -eq 0 ]; do
    sleep 15
    ALL_DNS_LIVE=1
    for DN_HOST in "${@:3}"; do
      # Liveness report may be stale, so verify that last contact from DN was recent.
      # Allow 4 failures in a row before giving up on getting the last contact time.
      GET_LAST_CONTACT="$GET_LAST_CONTACT_BASE print '$DN_HOST' in liveNodes and str(liveNodes['$DN_HOST']['lastContact']) or 'NOT_LIVE'"
      JMX_FAIL_COUNT=0
      while true; do
        LAST_CONTACT_SECONDS=`python -c "$ADD_IMPORTS $GET_LAST_CONTACT"`
        if [ $? -ne 0 ]; then
          JMX_FAIL_COUNT=$(($JMX_FAIL_COUNT+1))
          if [ $JMX_FAIL_COUNT -lt 4 ]; then
            echo "Failed to get last contact time for DataNode $DN_HOST from $NN_JMX. Retrying."
            sleep 15
          else
            echo "Failed to get last contact time for DataNode $DN_HOST from $NN_JMX 4 times in a row. Exiting."
            exit 1
          fi
        else
          break
        fi
      done

      if [ "NOT_LIVE" = $LAST_CONTACT_SECONDS ] || [ $LAST_CONTACT_SECONDS -gt 30 ]; then
        ALL_DNS_LIVE=0
        break
      fi
    done
  done
  echo "Success: DataNodes are live."
  exit 0
elif [ "monitor-offline" = "$1" ]; then
  # $2 is the NameNode's /jmx URL
  # $3 is the hostname of DataNode to check for offline (maintenance state).
  if [ $# -lt 3 ]; then
    echo "No DataNode hostname specified."
    exit 1
  fi

  NN_JMX="$2"
  DN_HOST="$3"

  ADD_IMPORTS="from cmf.url_util import urlopen_with_timeout; import simplejson; "
  GET_ADMIN_STATE="nnInfo = simplejson.loads(urlopen_with_timeout('$NN_JMX').read());"

  # The datanode can show up in LiveNodes or DeadNodes; both need to be checked.
  # It will show up in LiveNodes if the DN is still running.
  # It will show up in DeadNodes if the DN is no longer running.
  GET_LIVE_ADMIN_STATE="$GET_ADMIN_STATE liveNodes = simplejson.loads(nnInfo['beans'][0]['LiveNodes']); "
  GET_LIVE_ADMIN_STATE="$GET_LIVE_ADMIN_STATE print '$DN_HOST' in liveNodes and str(liveNodes['$DN_HOST']['adminState']) or 'NOT_PRESENT'"

  GET_DEAD_ADMIN_STATE="$GET_ADMIN_STATE deadNodes = simplejson.loads(nnInfo['beans'][0]['DeadNodes']); "
  GET_DEAD_ADMIN_STATE="$GET_DEAD_ADMIN_STATE print '$DN_HOST' in deadNodes and str(deadNodes['$DN_HOST']['adminState']) or 'NOT_PRESENT'"

  ADMIN_STATE=""
  while [ "$ADMIN_STATE" != "In Maintenance" ]; do
    sleep 15

    # Liveness report may be stale, so verify that last contact from DN was recent.
    # Allow 4 failures in a row before giving up on getting the admin state.
    JMX_FAIL_COUNT=0
    while true; do
      ADMIN_STATE=`python -c "$ADD_IMPORTS $GET_LIVE_ADMIN_STATE"`
      PY_STAT=$?
      if [ $PY_STAT -eq 0 ]; then
        if [ "$ADMIN_STATE" = "NOT_PRESENT" ]; then
          ADMIN_STATE=`python -c "$ADD_IMPORTS $GET_DEAD_ADMIN_STATE"`
          PY_STAT=$?
        fi
      fi

      if [ $PY_STAT -ne 0 ]; then
        JMX_FAIL_COUNT=$(($JMX_FAIL_COUNT+1))
        if [ $JMX_FAIL_COUNT -lt 4 ]; then
          echo "Failed to get admin state for DataNode $DN_HOST from $NN_JMX. Retrying."
          sleep 15
        else
          echo "Failed to get admin state for DataNode $DN_HOST from $NN_JMX 4 times in a row. Exiting."
          exit 1
        fi
      else
        break
      fi
    done
    if [ "$ADMIN_STATE" = "NOT_PRESENT" ]; then
      echo "DataNode $DN_HOST is no longer present"
      exit 1
    fi
  done
  echo "Success: $DN_HOST is ${ADMIN_STATE}"
  exit 0
elif [ "refresh-datanode" = "$1" ]; then
  if [ $# -lt 3 ]; then
    echo "No reconfigurable properties specified."
    exit 1
  fi

  DN_ADDR="$2"
  $HDFS_BIN --config $CONF_DIR dfsadmin -reconfig datanode "$DN_ADDR" start
  RET=$?
  if [ $RET -ne 0 ]; then
    echo "Unable to start reconfigure task on DataNode $DN_ADDR."
    exit $RET
  fi

  # Poll for completion of the reconfigure task. The task will usually complete
  # very quickly, so we sleep for one second at the top of the loop, then for
  # a longer time within the loop if the task has not finished.
  sleep 1
  RECONFIG_DONE=0
  FAIL_COUNT=0
  while [ $RECONFIG_DONE -eq 0 ]; do
    STATUS=`$HDFS_BIN --config $CONF_DIR dfsadmin -reconfig datanode "$DN_ADDR" status`
    # We allow 4 failures in a row before giving up on getting reconfig status,
    # to handle transient network issues.
    if [ $? -ne 0 ]; then
      FAIL_COUNT=$(($FAIL_COUNT+1))
      if [ $FAIL_COUNT -ge 4 ]; then
        echo "Failed to get status of reconfigure task on DataNode $DN_ADDR 4 times in a row. Exiting ... "
        exit 1
      else
        echo "Failed to get status of reconfigure task on DataNode $DN_ADDR. Retrying."
      fi
    else
      FAIL_COUNT=0 # Reset fail count
    fi

    echo "$STATUS" | grep "still running"
    RECONFIG_DONE=$?
    if [ $RECONFIG_DONE -eq 0 ]; then
        sleep 15
    fi
  done

  RET=0
  echo "$STATUS" | grep "finished at"
  if [ $? -ne 0 ]; then
    echo "Reconfigure task on DataNode $DN_ADDR failed to complete."
    RET=1
  else
    # Arguments $3 and following are names of reconfigurable properties.
    PATTERNS="$3"
    for PROP in "${@:4}"; do
      PATTERNS="$PATTERNS|$PROP"
    done

    echo "$STATUS" | grep "FAILED:" | grep -E "$PATTERNS"
    if [ $? -eq 0 ]; then
      echo "Failed to reconfigure one or more DataNode properties; see stderr log. Please revert the configuration change in Cloudera Manager."
      RET=1
    fi
  fi

  exit $RET
elif [ "mkdir" = "$1" ]; then
  if [ $# -ne 5 ]; then
    echo "mkdir not called with expected number of arguments. (4)"
    exit 1
  fi

  DIR=$2
  USER=$3
  GROUP=$4
  PERMS=$5

  if [ -z "$FAST_MKDIR" ]; then
    # Do mkdir the old way with separate hadoop calls
    RET=0
    $HDFS_BIN --config $CONF_DIR dfs -test -d "$DIR"
    if [ $? -eq 0 ]; then
      echo "Directory $DIR already exists."
    else
      echo "Directory $DIR does not currently exist."

      # Starting with CDH 4.1, HDFS won't automatically create the parent dir
      # if -p is not specified.
      PARENT_ARG=""
      if [ "$CDH_VERSION" -ge "4" ]; then
        PARENT_ARG="-p"
      fi
      $HDFS_BIN --config $CONF_DIR dfs -mkdir $PARENT_ARG "$DIR"
      if [ $? -eq 0 ]; then
        echo "Created directory $DIR."
      else
        echo "Unable to create directory $DIR; see stderr log."
        exit 1
      fi

      $HDFS_BIN --config $CONF_DIR dfs -chown "$USER":"$GROUP" "$DIR"
      if [ $? -eq 0 ]; then
        echo "Changed ownership of $DIR to $USER:$GROUP."
      else
        echo "Unable to change ownership of $DIR to $USER:$GROUP; see stderr log."
        RET=1
      fi
      $HDFS_BIN --config $CONF_DIR dfs -chmod "$PERMS" "$DIR"
      if [ $? -eq 0 ]; then
        echo "Changed permissions of $DIR to $PERMS."
      else
        echo "Unable to change permissions of $DIR to $PERMS; see stderr log."
        RET=1
      fi
    fi

    exit $RET
  else

    # Do new fast mdkir
    # MGMT_HOME existence is checked in cloudera-config.sh.
    # "*" is interpreted by Java, to mean all jars in dir.
    MGMT_CLASSPATH="${MGMT_HOME}/lib/*"

    # mkdir requires log4j file that doesn't contain CDH client classes
    # so a simple one is created here
    MKDIR_LOG4J_FILE="${CONF_DIR}/log4j_mkdir_$$.properties"
    rm -f ${MKDIR_LOG4J_FILE}
    cat << EOF > ${MKDIR_LOG4J_FILE}
# Basic, generic log4j configuration file for CreateHdfsDirUtil
# The default one in this directory 'log4j.properties' references classes
# that CreateHdfsDirUtil does not have access to.

# Root logger option
log4j.rootLogger=WARN, stdout

# Direct log messages to stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.Target=System.out
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} %-5p %c{1}:%L - %m%n
EOF

    MKDIR_JAVA_OPTS="${MKDIR_JAVA_OPTS} -Dlog4j.configuration=file://${MKDIR_LOG4J_FILE}"

    if [ "${JAVA_HOME}" != "" ]; then
      JAVA="${JAVA_HOME}/bin/java"
    fi

    MKDIR_CLASSPATH=${PWD}:${MGMT_CLASSPATH}

    MKDIR_FLAGS="--mgmt_home ${MGMT_HOME}"
    MKDIR_FLAGS="${MKDIR_FLAGS} --cdh_version ${CDH_VERSION}"
    MKDIR_FLAGS="${MKDIR_FLAGS} --conf_dir ${CONF_DIR}"
    if [ -n "${SCM_KERBEROS_PRINCIPAL}" ]; then
      MKDIR_FLAGS="${MKDIR_FLAGS} --use_kerberos"
      export SCM_KERBEROS_PRINCIPAL=${SCM_KERBEROS_PRINCIPAL}
    fi

    exec ${JAVA} ${MKDIR_JAVA_OPTS} -cp ${MKDIR_CLASSPATH} com.cloudera.cmf.cdhclient.common.hdfs.CreateHdfsDirUtil ${DIR} ${USER} ${GROUP} ${PERMS} ${MKDIR_FLAGS}
  fi
elif [ "nfs3" = "$1" ]; then
  # Check if portmap is running
  connect_to_port $NFSGATEWAY_PORTMAP_PORT
  if [ $? -eq 1 ]; then
    echo "No portmap or rpcbind service is running on this host. Please start portmap or rpcbind service before attempting to start the NFS Gateway role on this host."
    exit 1
  fi

  # Check if nfs service is already running
  connect_to_port $NFSGATEWAY_SERVER_PORT
  if [ $? -eq 0 ]; then
    echo "NFS service is already running on this host. Please stop the NFS service running on this host before attempting to start the NFS Gateway role."
   exit 1
  fi

  if [ -n "$PRIVILEGED_USER" ]; then
    export HADOOP_PRIVILEGED_NFS_USER=$PRIVILEGED_USER
    set_jsvc_home
    # The agent will spawn NFS Gateway as root:root but NFS eventually relinquishes
    # privileges to whatever system user HDFS service is configured with, so file
    # ownership must be reset.
    chown -R $PRIVILEGED_USER:$PRIVILEGED_GROUP $CONF_DIR
  fi
  export HADOOP_OPTS="$HADOOP_NFS3_OPTS $HADOOP_OPTS"
  exec $HDFS_BIN --config $CONF_DIR "$@"
elif [ "namenode" = "$1" -o \
       "secondarynamenode" = "$1" -o \
       "datanode" = "$1" ]; then
  # Set hadoop security and audit log appenders. These are set here instead
  # of being hardcoded in the log4j template because we only want the hadoop
  # daemons to use them.
  HADOOP_OPTS="-Dsecurity.audit.logger=$HADOOP_SECURITY_LOGGER $HADOOP_OPTS"
  export HADOOP_OPTS="-Dhdfs.audit.logger=$HADOOP_AUDIT_LOGGER $HADOOP_OPTS"

  # If we are doing a rolling upgrade, check if s/w version is different
  # from version in namenode data dir.
  if [ "namenode" = "$1" -a "rollingUpgrade" = "$2" ]; then
    VERSION_CHECK=`$HDFS_BIN --config $CONF_DIR namenode -metadataVersion`
    if [ $? -ne 0 ]; then
      echo "Failed to do HDFS metadata version check."
      exit 1
    fi
    # VERSION_CHECK has format like this:
    # HDFS Image Version: -55 Software format version: -59
    DATA_VERSION=`echo $VERSION_CHECK | cut -d ' ' -f 4`
    SOFTWARE_VERSION=`echo $VERSION_CHECK | cut -d ' ' -f 8`
    if [ "$DATA_VERSION" != "$SOFTWARE_VERSION" ]; then
      exec $HDFS_BIN --config $CONF_DIR namenode -rollingUpgrade started
    else
      # No version change, just start namenode as usual
      exec $HDFS_BIN --config $CONF_DIR namenode
    fi
  fi

  exec $HDFS_BIN --config $CONF_DIR "$@"
else
  exec $HDFS_BIN --config $CONF_DIR "$@"
fi
