#!/usr/bin/env bash
#
# 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
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# 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.
#

BINDIR=$(dirname "$0")
export PULSAR_HOME=`cd -P $BINDIR/..;pwd`
export BK_HOME=`cd -P $BINDIR/..;pwd`

# log directory
PULSAR_LOG_DIR=${PULSAR_LOG_DIR:-"$PULSAR_HOME/logs"}

DEFAULT_BROKER_CONF=$PULSAR_HOME/conf/broker.conf
DEFAULT_BOOKKEEPER_CONF=$PULSAR_HOME/conf/bookkeeper.conf
DEFAULT_ZK_CONF=$PULSAR_HOME/conf/zookeeper.conf
DEFAULT_CONFIGURATION_STORE_CONF=$PULSAR_HOME/conf/global_zookeeper.conf
DEFAULT_PROXY_CONF=$PULSAR_HOME/conf/proxy.conf
DEFAULT_STANDALONE_CONF=$PULSAR_HOME/conf/standalone.conf
DEFAULT_WEBSOCKET_CONF=$PULSAR_HOME/conf/websocket.conf
DEFAULT_LOG_CONF=$PULSAR_HOME/conf/log4j2.yaml
DEFAULT_PULSAR_PRESTO_CONF=${PULSAR_HOME}/conf/presto
DEFAULT_FUNCTIONS_LOG_CONF=$PULSAR_HOME/conf/functions_log4j2.xml

# functions related variables
FUNCTIONS_HOME=$PULSAR_HOME/pulsar-functions
DEFAULT_WORKER_CONF=$PULSAR_HOME/conf/functions_worker.yml
DEFAULT_JAVA_INSTANCE_JAR=$PULSAR_HOME/instances/java-instance.jar
JAVA_INSTANCE_JAR=${PULSAR_JAVA_INSTANCE_JAR:-"${DEFAULT_JAVA_INSTANCE_JAR}"}
DEFAULT_PY_INSTANCE_FILE=$PULSAR_HOME/instances/python-instance/python_instance_main.py
PY_INSTANCE_FILE=${PULSAR_PY_INSTANCE_FILE:-"${DEFAULT_PY_INSTANCE_FILE}"}
DEFAULT_FUNCTIONS_EXTRA_DEPS_DIR=$PULSAR_HOME/instances/deps
FUNCTIONS_EXTRA_DEPS_DIR=${PULSAR_FUNCTIONS_EXTRA_DEPS_DIR:-"${DEFAULT_FUNCTIONS_EXTRA_DEPS_DIR}"}
SQL_HOME=$PULSAR_HOME/pulsar-sql
PRESTO_HOME=${PULSAR_HOME}/lib/presto

pulsar_help() {
    cat <<EOF
Usage: pulsar <command>
where command is one of:

    broker              Run a broker server
    bookie              Run a bookie server
    zookeeper           Run a zookeeper server
    configuration-store Run a configuration-store server
    proxy               Run a pulsar proxy
    websocket           Run a web socket proxy server
    functions-worker    Run a functions worker server
    sql-worker          Run a sql worker server
    sql                 Run sql CLI
    standalone          Run a broker server with local bookies and local zookeeper
    autorecovery        Run an autorecovery service

    initialize-cluster-metadata     One-time metadata initialization
    delete-cluster-metadata         Delete a cluster's metadata
    initialize-transaction-coordinator-metadata     One-time transaction coordinator metadata initialization
    initialize-namespace     namespace initialization
    compact-topic       Run compaction against a topic
    zookeeper-shell     Open a ZK shell client
    broker-tool         CLI to operate a specific broker
    tokens              Utility to create authentication tokens
    version             Get the current version of pulsar

    help                This help message

or command is the full name of a class with a defined main() method.

Environment variables:
   FUNCTIONS_LOG_CONF            Function Log4j configuration file (default $DEFAULT_FUNCTIONS_LOG_CONF)
   PULSAR_LOG_CONF               Log4j configuration file (default $DEFAULT_LOG_CONF)
   PULSAR_BROKER_CONF            Configuration file for broker (default: $DEFAULT_BROKER_CONF)
   PULSAR_BOOKKEEPER_CONF        Configuration file for bookie (default: $DEFAULT_BOOKKEEPER_CONF)
   PULSAR_ZK_CONF                Configuration file for zookeeper (default: $DEFAULT_ZK_CONF)
   PULSAR_CONFIGURATION_STORE_CONF         Configuration file for global configuration store (default: $DEFAULT_CONFIGURATION_STORE_CONF)
   PULSAR_WEBSOCKET_CONF         Configuration file for websocket proxy (default: $DEFAULT_WEBSOCKET_CONF)
   PULSAR_PROXY_CONF             Configuration file for Pulsar proxy (default: $DEFAULT_PROXY_CONF)
   PULSAR_WORKER_CONF            Configuration file for functions worker (default: $DEFAULT_WORKER_CONF)
   PULSAR_STANDALONE_CONF        Configuration file for standalone (default: $DEFAULT_STANDALONE_CONF)
   PULSAR_PRESTO_CONF            Configuration directory for Pulsar Presto (default: $DEFAULT_PULSAR_PRESTO_CONF)
   PULSAR_EXTRA_OPTS             Extra options to be passed to the jvm
   PULSAR_EXTRA_CLASSPATH        Add extra paths to the pulsar classpath
   PULSAR_PID_DIR                Folder where the pulsar server PID file should be stored
   PULSAR_STOP_TIMEOUT           Wait time before forcefully kill the pulsar server instance, if the stop is not successful

These variable can also be set in conf/pulsar_env.sh
EOF
}

# if no args specified, show usage
if [ $# = 0 ]; then
    pulsar_help;
    exit 1;
fi

# get arguments
COMMAND=$1
shift

# Check bookkeeper env and load bkenv.sh
if [ -f "$PULSAR_HOME/conf/bkenv.sh" ]
then
    . "$PULSAR_HOME/conf/bkenv.sh"
fi

# Check pulsar env and load pulsar_env.sh
if [[ -f "$PULSAR_HOME/conf/pulsar_env.sh" && $COMMAND != "bookie" ]]
then
    . "$PULSAR_HOME/conf/pulsar_env.sh"
fi

# Check for the java to use
if [[ -z $JAVA_HOME ]]; then
    JAVA=$(which java)
    if [ $? != 0 ]; then
        echo "Error: JAVA_HOME not set, and no java executable found in $PATH." 1>&2
        exit 1
    fi
else
    JAVA=$JAVA_HOME/bin/java
fi

# exclude tests jar
RELEASE_JAR=`ls $PULSAR_HOME/pulsar-*.jar 2> /dev/null | grep -v tests | tail -1`
if [ $? == 0 ]; then
    PULSAR_JAR=$RELEASE_JAR
fi

# exclude tests jar
BUILT_JAR=`ls $PULSAR_HOME/pulsar-broker/target/pulsar-*.jar 2> /dev/null | grep -v tests | tail -1`
if [ $? != 0 ] && [ ! -e "$PULSAR_JAR" ]; then
    echo "\nCouldn't find pulsar jar.";
    echo "Make sure you've run 'mvn package'\n";
    exit 1;
elif [ -e "$BUILT_JAR" ]; then
    PULSAR_JAR=$BUILT_JAR
fi

#
# find the instance locations for pulsar-functions
#

# find the java instance location
if [ ! -f "${JAVA_INSTANCE_JAR}" ]; then
    # didn't find a released jar, then search the built jar
    BUILT_JAVA_INSTANCE_JAR="${FUNCTIONS_HOME}/runtime-all/target/java-instance.jar"
    if [ -z "${BUILT_JAVA_INSTANCE_JAR}" ]; then
        echo "\nCouldn't find pulsar-functions java instance jar.";
        echo "Make sure you've run 'mvn package'\n";
        exit 1;
    fi
    JAVA_INSTANCE_JAR=${BUILT_JAVA_INSTANCE_JAR}
fi

# find the python instance location
if [ ! -f "${PY_INSTANCE_FILE}" ]; then
    # didn't find a released python instance, then search the built python instance
    BUILT_PY_INSTANCE_FILE="${FUNCTIONS_HOME}/instance/target/python-instance/python_instance_main.py"
    if [ -z "${BUILT_PY_INSTANCE_FILE}" ]; then
        echo "\nCouldn't find pulsar-functions python instance.";
        echo "Make sure you've run 'mvn package'\n";
        exit 1;
    fi
    PY_INSTANCE_FILE=${BUILT_PY_INSTANCE_FILE}
fi

# find pulsar sql presto distribution location
check_presto_libraries() {
    if [ ! -d "${PRESTO_HOME}" ]; then

        BUILT_PRESTO_HOME="${SQL_HOME}/presto-distribution/target/pulsar-presto-distribution"
        if [ ! -d "${BUILT_PRESTO_HOME}" ]; then
            echo "\nCouldn't find presto distribution.";
            echo "Make sure you've run 'mvn package'\n";
            exit 1;
        fi
        PRESTO_HOME=${BUILT_PRESTO_HOME}
    fi
}

add_maven_deps_to_classpath() {
    MVN="mvn"
    if [ "$MAVEN_HOME" != "" ]; then
	    MVN=${MAVEN_HOME}/bin/mvn
    fi

    # Need to generate classpath from maven pom. This is costly so generate it
    # and cache it. Save the file into our target dir so a mvn clean will get
    # clean it up and force us create a new one.
    f="${PULSAR_HOME}/distribution/server/target/classpath.txt"
    if [ ! -f "${f}" ]
    then
    (
      cd "${PULSAR_HOME}"
      ${MVN} -pl distribution/server generate-sources &> /dev/null
    )
    fi
    PULSAR_CLASSPATH=${CLASSPATH}:`cat "${f}"`
}

if [ -d "$PULSAR_HOME/lib" ]; then
	  PULSAR_CLASSPATH=$PULSAR_CLASSPATH:$PULSAR_HOME/lib/*
else
    add_maven_deps_to_classpath
fi

if [ -z "$PULSAR_WORKER_CONF" ]; then
    PULSAR_WORKER_CONF=$DEFAULT_WORKER_CONF
fi

if [ -z "$PULSAR_BROKER_CONF" ]; then
    PULSAR_BROKER_CONF=$DEFAULT_BROKER_CONF
fi

if [ -z "$PULSAR_BOOKKEEPER_CONF" ]; then
    PULSAR_BOOKKEEPER_CONF=$DEFAULT_BOOKKEEPER_CONF
fi

if [ -z "$PULSAR_ZK_CONF" ]; then
    PULSAR_ZK_CONF=$DEFAULT_ZK_CONF
fi

if [ -z "$PULSAR_GLOBAL_ZK_CONF" ]; then
    PULSAR_GLOBAL_ZK_CONF=$DEFAULT_GLOBAL_ZK_CONF
fi

if [ -z "$PULSAR_CONFIGURATION_STORE_CONF" ]; then
    PULSAR_CONFIGURATION_STORE_CONF=$DEFAULT_CONFIGURATION_STORE_CONF
fi

if [ -z "$PULSAR_PROXY_CONF" ]; then
    PULSAR_PROXY_CONF=$DEFAULT_PROXY_CONF
fi

if [ -z "$PULSAR_WEBSOCKET_CONF" ]; then
    PULSAR_WEBSOCKET_CONF=$DEFAULT_WEBSOCKET_CONF
fi

if [ -z "$PULSAR_STANDALONE_CONF" ]; then
    PULSAR_STANDALONE_CONF=$DEFAULT_STANDALONE_CONF
fi

if [ -z "$PULSAR_LOG_CONF" ]; then
    PULSAR_LOG_CONF=$DEFAULT_LOG_CONF
fi

if [ -z "$PULSAR_PRESTO_CONF" ]; then
    PULSAR_PRESTO_CONF=$DEFAULT_PULSAR_PRESTO_CONF
fi

if [ -z "$FUNCTIONS_LOG_CONF"]; then
  FUNCTIONS_LOG_CONF=$DEFAULT_FUNCTIONS_LOG_CONF
fi

PULSAR_CLASSPATH="$PULSAR_JAR:$PULSAR_CLASSPATH:$PULSAR_EXTRA_CLASSPATH"
PULSAR_CLASSPATH="`dirname $PULSAR_LOG_CONF`:$PULSAR_CLASSPATH"
OPTS="$OPTS -Dlog4j.configurationFile=`basename $PULSAR_LOG_CONF`"

# Ensure we can read bigger content from ZK. (It might be
# rarely needed when trying to list many z-nodes under a
# directory)
OPTS="$OPTS -Djute.maxbuffer=10485760 -Djava.net.preferIPv4Stack=true"
# Enable TCP keepalive for all Zookeeper client connections
OPTS="$OPTS -Dzookeeper.clientTcpKeepAlive=true"

# Allow Netty to use reflection access
OPTS="$OPTS -Dio.netty.tryReflectionSetAccessible=true"
IS_JAVA_8=`$JAVA -version 2>&1 |grep version|grep '"1\.8'`
# Start --add-opens options
# '--add-opens' option is not supported in jdk8
if [[ -z "$IS_JAVA_8" ]]; then
  # BookKeeper: enable posix_fadvise usage and DirectMemoryCRC32Digest (https://github.com/apache/bookkeeper/pull/3234)
  OPTS="$OPTS --add-opens java.base/java.io=ALL-UNNAMED --add-opens java.base/java.util.zip=ALL-UNNAMED"
  # Netty: enable java.nio.DirectByteBuffer
  # https://github.com/netty/netty/blob/4.1/common/src/main/java/io/netty/util/internal/PlatformDependent0.java
  # https://github.com/netty/netty/issues/12265
  OPTS="$OPTS --add-opens java.base/java.nio=ALL-UNNAMED --add-opens java.base/jdk.internal.misc=ALL-UNNAMED"
  # netty.DnsResolverUtil
  # JvmDefaultGCMetricsLogger
  OPTS="$OPTS --add-opens java.base/sun.net=ALL-UNNAMED --add-opens java.management/sun.management=ALL-UNNAMED"
fi

OPTS="-cp $PULSAR_CLASSPATH $OPTS"

if [ $COMMAND == "bookie" ]; then
  # Pass BOOKIE_EXTRA_OPTS option defined in bkenv.sh
  OPTS="$OPTS $BOOKIE_MEM $BOOKIE_GC $BOOKIE_GC_LOG $BOOKIE_EXTRA_OPTS"
else
  OPTS="$OPTS $PULSAR_MEM $PULSAR_GC $PULSAR_GC_LOG $PULSAR_EXTRA_OPTS"
fi

PULSAR_LOG_APPENDER=${PULSAR_LOG_APPENDER:-"RoutingAppender"}
PULSAR_LOG_LEVEL=${PULSAR_LOG_LEVEL:-"info"}
PULSAR_LOG_ROOT_LEVEL=${PULSAR_LOG_ROOT_LEVEL:-"${PULSAR_LOG_LEVEL}"}
PULSAR_ROUTING_APPENDER_DEFAULT=${PULSAR_ROUTING_APPENDER_DEFAULT:-"Console"}
if [ ! -d "$PULSAR_LOG_DIR" ]; then
  mkdir -p "$PULSAR_LOG_DIR"
fi

#Configure log configuration system properties
OPTS="$OPTS -Dpulsar.log.appender=$PULSAR_LOG_APPENDER"
OPTS="$OPTS -Dpulsar.log.dir=$PULSAR_LOG_DIR"
OPTS="$OPTS -Dpulsar.log.level=$PULSAR_LOG_LEVEL"
OPTS="$OPTS -Dpulsar.log.root.level=$PULSAR_LOG_ROOT_LEVEL"
OPTS="$OPTS -Dpulsar.routing.appender.default=$PULSAR_ROUTING_APPENDER_DEFAULT"


# Configure log4j2 to disable servlet webapp detection so that Garbage free logging can be used
PULSAR_LOG4J_CONF=${PULSAR_LOG4J_CONF:-"-Dlog4j2.is.webapp=false  -Dlog4j2.contextSelector=org.apache.logging.log4j.core.async.AsyncLoggerContextSelector -Dlog4j2.enableThreadlocals=true -Dlog4j2.enableDirectEncoders=true"}
OPTS="$OPTS $PULSAR_LOG4J_CONF"

# Functions related logging
OPTS="$OPTS -Dpulsar.functions.process.container.log.dir=$PULSAR_LOG_DIR"
# instance
OPTS="$OPTS -Dpulsar.functions.java.instance.jar=${JAVA_INSTANCE_JAR}"
OPTS="$OPTS -Dpulsar.functions.python.instance.file=${PY_INSTANCE_FILE}"
OPTS="$OPTS -Dpulsar.functions.extra.dependencies.dir=${FUNCTIONS_EXTRA_DEPS_DIR}"
OPTS="$OPTS -Dpulsar.functions.instance.classpath=${PULSAR_CLASSPATH}"
OPTS="$OPTS -Dpulsar.functions.log.conf=${FUNCTIONS_LOG_CONF}"

ZK_OPTS=" -Dzookeeper.4lw.commands.whitelist=* -Dzookeeper.snapshot.trust.empty=true -Dzookeeper.tcpKeepAlive=true"

LOG4J2_SHUTDOWN_HOOK_DISABLED="-Dlog4j.shutdownHookEnabled=false"

# Adding pulsar metadata as a recognized provider
BK_METADATA_OPTIONS="-Dbookkeeper.metadata.bookie.drivers=org.apache.pulsar.metadata.bookkeeper.PulsarMetadataBookieDriver -Dbookkeeper.metadata.client.drivers=org.apache.pulsar.metadata.bookkeeper.PulsarMetadataClientDriver"
OPTS="$OPTS $BK_METADATA_OPTIONS"

#Change to PULSAR_HOME to support relative paths
cd "$PULSAR_HOME"
if [ $COMMAND == "broker" ]; then
    PULSAR_LOG_FILE=${PULSAR_LOG_FILE:-"pulsar-broker.log"}
    exec $JAVA $LOG4J2_SHUTDOWN_HOOK_DISABLED $OPTS -Dpulsar.log.file=$PULSAR_LOG_FILE org.apache.pulsar.PulsarBrokerStarter --broker-conf $PULSAR_BROKER_CONF $@
elif [ $COMMAND == "bookie" ]; then
    PULSAR_LOG_FILE=${PULSAR_LOG_FILE:-"bookkeeper.log"}
    exec $JAVA $OPTS -Dpulsar.log.file=$PULSAR_LOG_FILE org.apache.bookkeeper.server.Main --conf $PULSAR_BOOKKEEPER_CONF $@
elif [ $COMMAND == "zookeeper" ]; then
    PULSAR_LOG_FILE=${PULSAR_LOG_FILE:-"zookeeper.log"}
    exec $JAVA ${ZK_OPTS} $OPTS -Dpulsar.log.file=$PULSAR_LOG_FILE org.apache.zookeeper.server.quorum.QuorumPeerMain $PULSAR_ZK_CONF $@
elif [ $COMMAND == "global-zookeeper" ]; then
    PULSAR_LOG_FILE=${PULSAR_LOG_FILE:-"global-zookeeper.log"}
    # Allow global ZK to turn into read-only mode when it cannot reach the quorum
    OPTS="${OPTS} ${ZK_OPTS} -Dreadonlymode.enabled=true"
    exec $JAVA $OPTS -Dpulsar.log.file=$PULSAR_LOG_FILE org.apache.zookeeper.server.quorum.QuorumPeerMain $PULSAR_GLOBAL_ZK_CONF $@
elif [ $COMMAND == "configuration-store" ]; then
    PULSAR_LOG_FILE=${PULSAR_LOG_FILE:-"configuration-store.log"}
    # Allow global ZK to turn into read-only mode when it cannot reach the quorum
    OPTS="${OPTS} ${ZK_OPTS} -Dreadonlymode.enabled=true"
    exec $JAVA $OPTS -Dpulsar.log.file=$PULSAR_LOG_FILE org.apache.zookeeper.server.quorum.QuorumPeerMain $PULSAR_CONFIGURATION_STORE_CONF $@
elif [ $COMMAND == "proxy" ]; then
    PULSAR_LOG_FILE=${PULSAR_LOG_FILE:-"pulsar-proxy.log"}
    exec $JAVA $OPTS -Dpulsar.log.file=$PULSAR_LOG_FILE org.apache.pulsar.proxy.server.ProxyServiceStarter --config $PULSAR_PROXY_CONF $@
elif [ $COMMAND == "websocket" ]; then
    PULSAR_LOG_FILE=${PULSAR_LOG_FILE:-"pulsar-websocket.log"}
    exec $JAVA $OPTS -Dpulsar.log.file=$PULSAR_LOG_FILE org.apache.pulsar.websocket.service.WebSocketServiceStarter $PULSAR_WEBSOCKET_CONF $@
elif [ $COMMAND == "functions-worker" ]; then
    PULSAR_LOG_FILE=${PULSAR_LOG_FILE:-"pulsar-functions-worker.log"}
    exec $JAVA $OPTS -Dpulsar.log.file=$PULSAR_LOG_FILE org.apache.pulsar.functions.worker.FunctionWorkerStarter -c $PULSAR_WORKER_CONF $@
elif [ $COMMAND == "standalone" ]; then
    PULSAR_LOG_FILE=${PULSAR_LOG_FILE:-"pulsar-standalone.log"}
    exec $JAVA $LOG4J2_SHUTDOWN_HOOK_DISABLED $OPTS ${ZK_OPTS} -Dpulsar.log.file=$PULSAR_LOG_FILE -Dpulsar.config.file=$PULSAR_STANDALONE_CONF org.apache.pulsar.PulsarStandaloneStarter $@
elif [ ${COMMAND} == "autorecovery" ]; then
    PULSAR_LOG_FILE=${PULSAR_LOG_FILE:-"pulsar-autorecovery.log"}
    exec $JAVA $OPTS -Dpulsar.log.file=$PULSAR_LOG_FILE org.apache.bookkeeper.replication.AutoRecoveryMain --conf $PULSAR_BOOKKEEPER_CONF $@
elif [ $COMMAND == "initialize-cluster-metadata" ]; then
    exec $JAVA $OPTS org.apache.pulsar.PulsarClusterMetadataSetup $@
elif [ $COMMAND == "delete-cluster-metadata" ]; then
    exec $JAVA $OPTS org.apache.pulsar.PulsarClusterMetadataTeardown $@
elif [ $COMMAND == "initialize-transaction-coordinator-metadata" ]; then
    exec $JAVA $OPTS org.apache.pulsar.PulsarTransactionCoordinatorMetadataSetup $@
elif [ $COMMAND == "initialize-namespace" ]; then
    exec $JAVA $OPTS org.apache.pulsar.PulsarInitialNamespaceSetup $@
elif [ $COMMAND == "zookeeper-shell" ]; then
    exec $JAVA $OPTS org.apache.zookeeper.ZooKeeperMain $@
elif [ $COMMAND == "broker-tool" ]; then
    exec $JAVA $OPTS org.apache.pulsar.broker.tools.BrokerTool $@
elif [ $COMMAND == "compact-topic" ]; then
    exec $JAVA $OPTS org.apache.pulsar.compaction.CompactorTool --broker-conf $PULSAR_BROKER_CONF $@
elif [ $COMMAND == "sql" ]; then
    check_presto_libraries
    exec $JAVA -cp "${PRESTO_HOME}/lib/*" io.prestosql.cli.Presto --server localhost:8081 "${@}"
elif [ $COMMAND == "sql-worker" ]; then
    check_presto_libraries
    exec python3 ${PRESTO_HOME}/bin/launcher.py --etc-dir ${PULSAR_PRESTO_CONF} "${@}"
elif [ $COMMAND == "tokens" ]; then
    exec $JAVA $OPTS org.apache.pulsar.utils.auth.tokens.TokensCliUtils $@
elif [ $COMMAND == "version" ]; then
    exec $JAVA $OPTS org.apache.pulsar.PulsarVersionStarter $@
elif [ $COMMAND == "help" -o $COMMAND == "--help" -o $COMMAND == "-h" ]; then
    pulsar_help;
else
    echo ""
    echo "-- Invalid command '$COMMAND' -- Use '$0 help' to get a list of valid commands"
    echo ""
    exit 1
fi