#!/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

#replace {{PID}} in HeapDumpPath option
export HADOOP_TASKTRACKER_OPTS=$(replace_pid $HADOOP_TASKTRACKER_OPTS)
export HADOOP_JOBTRACKER_OPTS=$(replace_pid $HADOOP_JOBTRACKER_OPTS)

if [ "$CDH_VERSION" -ge "4" ]; then
  export HADOOP_HOME=$CDH_MR1_HOME
  ADD_TO_CP=""
  ADD_TO_CP="$CDH_HADOOP_HOME/*:$ADD_TO_CP"
  ADD_TO_CP="$CDH_HADOOP_HOME/lib/*:$ADD_TO_CP"
  ADD_TO_CP="$CDH_HDFS_HOME/*:$ADD_TO_CP"
  ADD_TO_CP="$CDH_HDFS_HOME/lib/*:$ADD_TO_CP"
  export HADOOP_CLASSPATH="$ADD_TO_CP:$HADOOP_CLASSPATH"
  echo "using $HADOOP_CLASSPATH as HADOOP_CLASSPATH"

  # 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
  export HADOOP_HOME=$CDH_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 $HADOOP_HOME as HADOOP_HOME"
echo "using $CONF_DIR as HADOOP_CONF_DIR"

set_hadoop_classpath

# debug
set -x

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

# Make sure topology.py is executable
make_scripts_executable

acquire_kerberos_tgt mapred.keytab

export HADOOP_BIN=$HADOOP_HOME/bin/hadoop

if [ $1 = "jobtracker" -o $1 = "jobtrackerha" ]; then
  if [ z$ADD_CAPACITY_SCHEDULER = z"true" ]; then
    ADD_TO_CP=`find $HADOOP_HOME/contrib/capacity-scheduler -name 'hadoop-capacity-scheduler-[0-9]*.jar'`
    NUM_LINES=`echo "$ADD_TO_CP"|wc -l`
    if [ "$NUM_LINES" -gt "1" ]; then
      echo "Warning: More than one jar found in $HADOOP_HOME/contrib/capacity-scheduler, using first matching jar"
      ADD_TO_CP=`echo "$ADD_TO_CP"|head -1`
    fi
    ADD_TO_CP=`echo "$ADD_TO_CP" | tr -d "\n"`
    if [ -z $HADOOP_CLASSPATH ]; then
      export HADOOP_CLASSPATH="$ADD_TO_CP"
    else
      export HADOOP_CLASSPATH="$ADD_TO_CP:$HADOOP_CLASSPATH"
    fi
    env | grep HADOOP_CLASSPATH
  fi
elif [ $1 = "tasktracker" ]; then
  # Children need to be able to read the Hadoop config
  # files, so make those world-readable
  chmod 0644 $CONF_DIR/mapred-site.xml $CONF_DIR/hdfs-site.xml $CONF_DIR/core-site.xml $CONF_DIR/log4j.properties
  if [ "$CDH_VERSION" -ge "5" ]; then
    chmod 0644 $CONF_DIR/ssl-client.xml
  fi
elif [ $1 = "RefreshQueuesAndNodes" ]; then
  # If JT is in HA mode, we need to issue command only to active JT
  if [ -n $2 ] && [ $2 = "true" ]; then
    # find if JT is in standby mode
    HASTATE="$($HADOOP_BIN --config $CONF_DIR mrhaadmin -getServiceState $3)"
    if [[ $HASTATE =~ "standby" ]]; then
      # Standby JT have no state to refresh, exit successfully
      cm_success_and_exit JOBTRACKER_IN_STANDBY_MODE
    fi
  fi

  # If we are sending a refreshAll command to mradmin
  # it means we want to refresh queues and nodes
  RET=0
  $HADOOP_BIN --config $CONF_DIR mradmin -refreshQueues
  RET=$?
  if [ $RET -eq 0 ]; then
    echo "Refresh queues worked successfully"
    $HADOOP_BIN --config $CONF_DIR mradmin -refreshNodes
    RET=$?
    if [ $RET -eq 0 ]; then
      echo "Refresh nodes worked successfully"
    else
      echo "Refresh nodes failed"
    fi
  else
    echo "Refresh queues failed. Skipping refresh nodes."
  fi
  exit $RET
elif [ $1 = "jtRpcWait" ]; then
  # Wait for jobtracker to start responding to RPCs, by checking that we can
  # open the URL specified in $2.
  WAIT=1
  ADD_IMPORTS="from cmf.url_util import urlopen_with_timeout;"
  while [ $WAIT -ne 0 ]; do
    sleep 15
    python -c "$ADD_IMPORTS urlopen_with_timeout('$2')"
    WAIT=$?
  done
  exit 0
fi

exec $HADOOP_BIN --config $CONF_DIR "$@"
