echo

echo "--------------------------------------------------------------------------------------------"

echo "This example runs on Confluent Cloud, sign up here:"

echo

echo " https://www.confluent.io/confluent-cloud/tryfree/"

echo

echo "The example uses real Confluent Cloud resources that may be billable, including connectors"

echo "and ksqlDB applications that may have hourly charges. The end of this script shows a command"

echo "you can run to destroy all the cloud resources, and you should verify they are destroyed."

echo

echo "You may be billed for the Confluent Cloud resources until you destroy them."

echo "--------------------------------------------------------------------------------------------"

echo

read -p "Do you still want to run this script? [y/n] " -n 1 -r

echo

if [[ ! $REPLY =~ ^[Yy]$ ]]

then

exit 1

fi

return 0

if [[ $(type expect 2>&1) =~ "not found" ]]; then

echo "'expect' is not found. Install 'expect' and try again"

exit 1

fi

return 0

if [[ $(type confluent 2>&1) =~ "not found" ]]; then

echo "'confluent' is not found. Install the Confluent CLI (https://docs.confluent.io/confluent-cli/current/install.html) and try again."

exit 1

fi

ccloud::validate_cli_installed || exit 1

if [[ -z $(confluent version 2>&1 | grep "Go") ]]; then

echo "This example requires the new Confluent CLI. Please update your version and try again."

exit 1

fi

return 0

ccloud::validate_cli_v2 || exit 1

if [[ "$(confluent kafka cluster list 2>&1)" =~ "confluent login" ]]; then

echo

echo "ERROR: Not logged into Confluent Cloud."

echo "Log in with the command 'confluent login --save' before running the example. The '--save' argument saves your Confluent Cloud user login credentials or refresh token (in the case of SSO) to the local netrc file."

exit 1

fi

return 0

ccloud::validate_cli_installed || exit 1

CLI_VERSION=$(ccloud::get_version_cli)

if ccloud::version_gt $CLI_MIN_VERSION $CLI_VERSION; then

echo "confluent version ${CLI_MIN_VERSION} or greater is required. Current version: ${CLI_VERSION}"

echo "To update, follow: https://docs.confluent.io/confluent-cli/current/migrate.html"

exit 1

fi

if [[ $(type psql 2>&1) =~ "not found" ]]; then

echo "psql is not found. Install psql and try again"

exit 1

fi

return 0

if [[ $(type aws 2>&1) =~ "not found" ]]; then

echo "AWS CLI is not found. Install AWS CLI and try again"

exit 1

fi

return 0

version_major=$(aws --version 2>&1 | awk -F/ '{print $2;}' | head -c 1)

if [[ "$version_major" -eq 2 ]]; then

echo "2"

else

echo "1"

fi

return 0

if [[ $(type gsutil 2>&1) =~ "not found" ]]; then

echo "Google Cloud gsutil is not found. Install Google Cloud gsutil and try again"

exit 1

fi

return 0

if [[ $(type az 2>&1) =~ "not found" ]]; then

echo "Azure CLI is not found. Install Azure CLI and try again"

exit 1

fi

return 0

config=$1

source $config

if [[ "$DATA_SOURCE" == "kinesis" ]]; then

ccloud::validate_aws_cli_installed || exit 1

if [[ -z "$KINESIS_REGION" || -z "$AWS_PROFILE" ]]; then

echo "ERROR: DATA_SOURCE=kinesis, but KINESIS_REGION or AWS_PROFILE is not set. Please set these parameters in config/demo.cfg and try again."

exit 1

fi

aws kinesis list-streams --profile $AWS_PROFILE --region $KINESIS_REGION > /dev/null \

|| { echo "Could not run 'aws kinesis list-streams'. Check credentials and run again." ; exit 1; }

elif [[ "$DATA_SOURCE" == "rds" ]]; then

ccloud::validate_aws_cli_installed || exit 1

if [[ -z "$RDS_REGION" || -z "$AWS_PROFILE" ]]; then

echo "ERROR: DATA_SOURCE=rds, but RDS_REGION or AWS_PROFILE is not set. Please set these parameters in config/demo.cfg and try again."

exit 1

fi

aws rds describe-db-instances --profile $AWS_PROFILE --region $RDS_REGION > /dev/null \

|| { echo "Could not run 'aws rds describe-db-instances'. Check credentials and run again." ; exit 1; }

else

echo "Cloud source $cloudsource is not valid. Must be one of [kinesis|rds]."

exit 1

fi

return 0

config=$1

source $config

storage=$DESTINATION_STORAGE

if [[ "$storage" == "s3" ]]; then

ccloud::validate_aws_cli_installed || exit 1

ccloud::validate_credentials_s3 $S3_PROFILE $S3_BUCKET || exit 1

aws s3api list-buckets --profile $S3_PROFILE --region $STORAGE_REGION > /dev/null \

|| { echo "Could not run 'aws s3api list-buckets'. Check credentials and run again." ; exit 1; }

elif [[ "$storage" == "gcs" ]]; then

ccloud::validate_gsutil_installed || exit 1

ccloud::validate_credentials_gcp $GCS_CREDENTIALS_FILE $GCS_BUCKET || exit 1

elif [[ "$storage" == "az" ]]; then

ccloud::validate_az_installed || exit 1

ccloud::validate_credentials_az $AZBLOB_STORAGE_ACCOUNT $AZBLOB_CONTAINER || exit 1

else

echo "Storage destination $storage is not valid. Must be one of [s3|gcs|az]."

exit 1

fi

return 0

GCS_CREDENTIALS_FILE=$1

GCS_BUCKET=$2

if [[ -z "$GCS_CREDENTIALS_FILE" || -z "$GCS_BUCKET" ]]; then

echo "ERROR: DESTINATION_STORAGE=gcs, but GCS_CREDENTIALS_FILE or GCS_BUCKET is not set. Please set these parameters in config/demo.cfg and try again."

exit 1

fi

gcloud auth activate-service-account --key-file $GCS_CREDENTIALS_FILE || {

echo "ERROR: Cannot activate service account with key file $GCS_CREDENTIALS_FILE. Verify your credentials and try again."

exit 1

}

# Create JSON-formatted string of the GCS credentials

export GCS_CREDENTIALS=$(python ./stringify-gcp-credentials.py $GCS_CREDENTIALS_FILE)

# Remove leading and trailing double quotes, otherwise connector creation from CLI fails

GCS_CREDENTIALS=$(echo "${GCS_CREDENTIALS:1:${#GCS_CREDENTIALS}-2}")

return 0

AZBLOB_STORAGE_ACCOUNT=$1

AZBLOB_CONTAINER=$2

if [[ -z "$AZBLOB_STORAGE_ACCOUNT" || -z "$AZBLOB_CONTAINER" ]]; then

echo "ERROR: DESTINATION_STORAGE=az, but AZBLOB_STORAGE_ACCOUNT or AZBLOB_CONTAINER is not set. Please set these parameters in config/demo.cfg and try again."

exit 1

fi

if [[ "$AZBLOB_STORAGE_ACCOUNT" == "default" ]]; then

echo "ERROR: Azure Blob storage account name cannot be 'default'. Verify the value of the storage account name (did you create one?) in config/demo.cfg, as specified by the parameter AZBLOB_STORAGE_ACCOUNT, and try again."

exit 1

fi

exists=$(az storage account check-name --name $AZBLOB_STORAGE_ACCOUNT | jq -r .reason)

if [[ "$exists" != "AlreadyExists" ]]; then

echo "ERROR: Azure Blob storage account name $AZBLOB_STORAGE_ACCOUNT does not exist. Check the value of AZBLOB_STORAGE_ACCOUNT in config/demo.cfg and try again."

exit 1

fi

export AZBLOB_ACCOUNT_KEY=$(az storage account keys list --account-name $AZBLOB_STORAGE_ACCOUNT | jq -r '.[0].value')

if [[ "$AZBLOB_ACCOUNT_KEY" == "" ]]; then

echo "ERROR: Cannot get the key for Azure Blob storage account name $AZBLOB_STORAGE_ACCOUNT. Check the value of AZBLOB_STORAGE_ACCOUNT in config/demo.cfg, and your key, and try again."

exit 1

fi

return 0

S3_PROFILE=$1

S3_BUCKET=$2

if [[ -z "$S3_PROFILE" || -z "$S3_BUCKET" ]]; then

echo "ERROR: DESTINATION_STORAGE=s3, but S3_PROFILE or S3_BUCKET is not set. Please set these parameters in config/demo.cfg and try again."

exit 1

fi

aws configure get aws_access_key_id --profile $S3_PROFILE 1>/dev/null || {

echo "ERROR: Cannot determine aws_access_key_id from S3_PROFILE=$S3_PROFILE. Verify your credentials and try again."

exit 1

}

aws configure get aws_secret_access_key --profile $S3_PROFILE 1>/dev/null || {

echo "ERROR: Cannot determine aws_secret_access_key from S3_PROFILE=$S3_PROFILE. Verify your credentials and try again."

exit 1

}

return 0

auth=$1

sr_endpoint=$2

curl --silent -u $auth $sr_endpoint > /dev/null || {

echo "ERROR: Could not validate credentials to Confluent Cloud Schema Registry. Please troubleshoot"

exit 1

}

echo "Validated credentials to Confluent Cloud Schema Registry at $sr_endpoint"

return 0

SERVICE_ACCOUNT_ID=$1

ENVIRONMENT_NAME_PREFIX=${ENVIRONMENT_NAME_PREFIX:-"ccloud-stack-$SERVICE_ACCOUNT_ID"}

local environment_id=$(confluent environment list -o json | jq -r 'map(select(.name | startswith("'"$ENVIRONMENT_NAME_PREFIX"'"))) | .[].id')

echo $environment_id

return 0

ENVIRONMENT_NAME=$1

OUTPUT=$(confluent environment create $ENVIRONMENT_NAME -o json)

(($? != 0)) && { echo "ERROR: Failed to create environment $ENVIRONMENT_NAME. Please troubleshoot and run again"; exit 1; }

ENVIRONMENT=$(echo "$OUTPUT" | jq -r ".id")

confluent environment use $ENVIRONMENT &>/dev/null

echo $ENVIRONMENT

return 0

CLUSTER_NAME=$1

CLUSTER_CLOUD=$2

CLUSTER_REGION=$3

local FOUND_CLUSTER=$(confluent kafka cluster list -o json | jq -c -r '.[] | select((.name == "'"$CLUSTER_NAME"'") and (.provider == "'"$CLUSTER_CLOUD"'") and (.region == "'"$CLUSTER_REGION"'"))')

[[ ! -z "$FOUND_CLUSTER" ]] && {

echo "$FOUND_CLUSTER" | jq -r .id

return 0

} || {

return 1

}

CLUSTER_NAME=$1

CLUSTER_CLOUD=$2

CLUSTER_REGION=$3

OUTPUT=$(confluent kafka cluster create "$CLUSTER_NAME" --cloud $CLUSTER_CLOUD --region $CLUSTER_REGION --output json 2>&1)

(($? != 0)) && { echo "$OUTPUT"; exit 1; }

CLUSTER=$(echo "$OUTPUT" | jq -r .id)

confluent kafka cluster use $CLUSTER 2>/dev/null

echo $CLUSTER

return 0

CLUSTER_NAME=$1

CLUSTER_CLOUD=$2

CLUSTER_REGION=$3

CLUSTER_ID=$(ccloud::find_cluster $CLUSTER_NAME $CLUSTER_CLOUD $CLUSTER_REGION)

if [ $? -eq 0 ]

then

confluent kafka cluster use $CLUSTER_ID

echo $CLUSTER_ID

else

OUTPUT=$(ccloud::create_and_use_cluster "$CLUSTER_NAME" "$CLUSTER_CLOUD" "$CLUSTER_REGION")

(($? != 0)) && { echo "$OUTPUT"; exit 1; }

echo "$OUTPUT"

fi

return 0

SERVICE_NAME=$1

CCLOUD_EMAIL=$(confluent prompt -f '%u')

OUTPUT=$(confluent iam service-account create $SERVICE_NAME --description "SA for $EXAMPLE run by $CCLOUD_EMAIL" -o json)

SERVICE_ACCOUNT_ID=$(echo "$OUTPUT" | jq -r ".id")

echo $SERVICE_ACCOUNT_ID

return 0

SERVICE_ACCOUNT_ID=$(confluent kafka cluster describe -o json | jq -r '.name' | awk -F'-' '{print $4 "-" $5;}')

echo $SERVICE_ACCOUNT_ID

return 0

SCHEMA_REGISTRY_CLOUD=$1

SCHEMA_REGISTRY_GEO=$2

OUTPUT=$(confluent schema-registry cluster enable --cloud $SCHEMA_REGISTRY_CLOUD --geo $SCHEMA_REGISTRY_GEO -o json)

SCHEMA_REGISTRY=$(echo "$OUTPUT" | jq -r ".id")

echo $SCHEMA_REGISTRY

return 0

SERVICE_ACCOUNT_ID=$1

RESOURCE=$2

local FOUND_CRED=$(confluent api-key list -o json | jq -c -r 'map(select((.resource_id == "'"$RESOURCE"'") and (.owner_resource_id == "'"$SERVICE_ACCOUNT_ID"'")))')

local FOUND_COUNT=$(echo "$FOUND_CRED" | jq 'length')

[[ $FOUND_COUNT -ne 0 ]] && {

echo "$FOUND_CRED" | jq -r '.[0].key'

return 0

} || {

return 1

}

SERVICE_ACCOUNT_ID=$1

RESOURCE=$2

OUTPUT=$(confluent api-key create --service-account $SERVICE_ACCOUNT_ID --resource $RESOURCE -o json)

API_KEY_SA=$(echo "$OUTPUT" | jq -r ".api_key")

API_SECRET_SA=$(echo "$OUTPUT" | jq -r ".api_secret")

echo "${API_KEY_SA}:${API_SECRET_SA}"

return 0

SERVICE_ACCOUNT_ID=$1

RESOURCE=$2

local KEY=$(ccloud::find_credentials_resource $SERVICE_ACCOUNT_ID $RESOURCE)

[[ -z $KEY ]] && {

ccloud::create_credentials_resource $SERVICE_ACCOUNT_ID $RESOURCE || exit 1

} || {

echo "$KEY:"; # the secret cannot be retrieved from a found key, caller needs to handle this

return 0

}

KSQLDB_NAME=$1

CLUSTER=$2

local FOUND_APP=$(confluent ksql cluster list -o json | jq -c -r 'map(select((.name == "'"$KSQLDB_NAME"'") and (.kafka == "'"$CLUSTER"'")))')

local FOUND_COUNT=$(echo "$FOUND_APP" | jq 'length')

[[ $FOUND_COUNT -ne 0 ]] && {

echo "$FOUND_APP" | jq -r '.[].id'

return 0

} || {

return 1

}

KSQLDB_NAME=$1

CLUSTER=$2

SERVICE_ACCOUNT_ID=$3

KSQLDB=$(confluent ksql cluster create --cluster $CLUSTER --credential-identity $SERVICE_ACCOUNT_ID --csu 1 -o json "$KSQLDB_NAME" | jq -r ".id")

echo $KSQLDB

return 0

KSQLDB_NAME=$1

CLUSTER=$2

# colon deliminated credentials (APIKEY:APISECRET)

local ksqlDB_kafka_creds=$3

APP_ID=$(ccloud::find_ksqldb_app $KSQLDB_NAME $CLUSTER)

if [ $? -eq 0 ]

then

echo $APP_ID

else

ccloud::create_ksqldb_app "$KSQLDB_NAME" "$CLUSTER" "$ksqlDB_kafka_creds"

fi

return 0

SERVICE_ACCOUNT_ID=$1

ENVIRONMENT=$2

# Setting default QUIET=false to surface potential errors

QUIET="${QUIET:-false}"

[[ $QUIET == "true" ]] &&

local REDIRECT_TO="/dev/null" ||

local REDIRECT_TO="/dev/tty"

echo "Adding role-binding to ${SERVICE_ACCOUNT_ID} on ${ENVIRONMENT}"

confluent iam rbac role-binding create --principal User:${SERVICE_ACCOUNT_ID} --role EnvironmentAdmin --environment ${ENVIRONMENT} -o json &>"$REDIRECT_TO"

echo -e "\nWaiting for role-binding to propagate\n"

sleep 30

confluent iam rbac role-binding list --principal User:${SERVICE_ACCOUNT_ID} --role EnvironmentAdmin --environment ${ENVIRONMENT} -o json &>"$REDIRECT_TO"

return 0

[ -z "$1" ] && {

echo "ccloud::validate_ccloud_config expects one parameter (configuration file with Confluent Cloud connection information)"

exit 1

}

local cfg_file="$1"

ccloud::validate_logged_in_cli

ccloud::validate_ccloud_config $cfg_file

ccloud::generate_configs $cfg_file

source delta_configs/env.delta

ccloud::set_cli_from_env_params

return 0

# Check minimum parameters

if [[ -z "$ENVIRONMENT_ID" || \

"$ENVIRONMENT_ID" == -1 || \

-z "$KAFKA_CLUSTER_ID" || \

"$KAFKA_CLUSTER_ID" == -1 || \

-z "$CLOUD_KEY" || \

-z "$CLOUD_SECRET" ]]; then

echo "ERROR: Missing at least one environment parameter. Please troubleshoot and run again."

return 1

fi

confluent environment use $ENVIRONMENT_ID

confluent kafka cluster use $KAFKA_CLUSTER_ID

confluent api-key store "$CLOUD_KEY" "$CLOUD_SECRET" --resource ${KAFKA_CLUSTER_ID}

confluent api-key use "$CLOUD_KEY" --resource ${KAFKA_CLUSTER_ID}

return 0

[ -z "$1" ] && {

echo "ccloud::validate_ccloud_config expects one parameter (configuration file with Confluent Cloud connection information)"

exit 1

}

local cfg_file="$1"

local bootstrap=$(grep "bootstrap\.servers" "$cfg_file" | cut -d'=' -f2-)

[ -z "$bootstrap" ] && {

echo "ERROR: Cannot read the 'bootstrap.servers' key-value pair from $cfg_file."

exit 1;

}

return 0;

[ -z "$1" ] && {

echo "ccloud::validate_ksqldb_up expects one parameter (ksqldb endpoint)"

exit 1

}

[ $# -gt 1 ] && echo "WARN: ccloud::validate_ksqldb_up function expects one parameter"

local ksqldb_endpoint=$1

ccloud::validate_logged_in_cli || exit 1

local ksqldb_meta=$(confluent ksql cluster list -o json | jq -r 'map(select(.endpoint == "'"$ksqldb_endpoint"'")) | .[]')

local ksqldb_appid=$(echo "$ksqldb_meta" | jq -r '.id')

if [[ "$ksqldb_appid" == "" ]]; then

echo "ERROR: Confluent Cloud ksqlDB endpoint $ksqldb_endpoint is not found. Provision a ksqlDB cluster via the Confluent Cloud UI and add the configuration parameter ksql.endpoint and ksql.basic.auth.user.info into your Confluent Cloud configuration file at $ccloud_config_file and try again."

exit 1

fi

local ksqldb_status=$(echo "$ksqldb_meta" | jq -r '.status')

if [[ $ksqldb_status != "PROVISIONED" ]]; then

echo "ERROR: Confluent Cloud ksqlDB endpoint $ksqldb_endpoint with id $ksqlDBAppId is not in PROVISIONED state. Troubleshoot and try again."

exit 1

fi

return 0

AZBLOB_STORAGE_ACCOUNT=$1

if [[ "$AZBLOB_STORAGE_ACCOUNT" == "default" ]]; then

echo "ERROR: Azure Blob storage account name cannot be 'default'. Verify the value of the storage account name (did you create one?) in config/demo.cfg, as specified by the parameter AZBLOB_STORAGE_ACCOUNT, and try again."

exit 1

fi

exists=$(az storage account check-name --name $AZBLOB_STORAGE_ACCOUNT | jq -r .reason)

if [[ "$exists" != "AlreadyExists" ]]; then

echo "ERROR: Azure Blob storage account name $AZBLOB_STORAGE_ACCOUNT does not exist. Check the value of STORAGE_PROFILE in config/demo.cfg and try again."

exit 1

fi

export AZBLOB_ACCOUNT_KEY=$(az storage account keys list --account-name $AZBLOB_STORAGE_ACCOUNT | jq -r '.[0].value')

if [[ "$AZBLOB_ACCOUNT_KEY" == "" ]]; then

echo "ERROR: Cannot get the key for Azure Blob storage account name $AZBLOB_STORAGE_ACCOUNT. Check the value of STORAGE_PROFILE in config/demo.cfg, and your key, and try again."

exit 1

fi

return 0

ksqldb_endpoint=$1

ccloud_config_file=$2

credentials=$3

response=$(curl ${ksqldb_endpoint}/info \

if [[ "$response" =~ "Unauthorized" ]]; then

echo "ERROR: Authorization failed to the ksqlDB cluster. Check your ksqlDB credentials set in the configuration parameter ksql.basic.auth.user.info in your Confluent Cloud configuration file at $ccloud_config_file and try again."

exit 1

fi

echo "Validated credentials to Confluent Cloud ksqlDB at $ksqldb_endpoint"

return 0

file=$1

echo -e "\nCreating connector from $file\n"

# About the Confluent CLI command 'confluent connect cluster create':

# - Typical usage of this CLI would be 'confluent connect cluster create --config <filename>'

# - However, in this example, the connector's configuration file contains parameters that need to be first substituted

# so the CLI command includes eval and heredoc.

# - The '-vvv' is added for verbose output

confluent connect cluster create -vvv --config-file <(eval "cat <<EOF

if [[ $? != 0 ]]; then

echo "ERROR: Exit status was not 0 while creating connector from $file. Please troubleshoot and try again"

exit 1

fi

return 0

filename=$1

maxWait=$2

connectorName=$(cat $filename | jq -r .name)

echo "Waiting up to $maxWait seconds for connector $filename ($connectorName) to be RUNNING"

ccloud::retry $maxWait ccloud::validate_connector_up $connectorName || exit 1

echo "Connector $filename ($connectorName) is RUNNING"

return 0

KSQLDB_ENDPOINT=$1

STATUS=$(confluent ksql cluster list -o json | jq -r 'map(select(.endpoint == "'"$KSQLDB_ENDPOINT"'")) | .[].status' | grep PROVISIONED)

if [[ "$STATUS" == "" ]]; then

return 1

fi

return 0

topic=$1

confluent kafka topic describe $topic &>/dev/null

return $?

subject=$1

sr_url=$2

sr_credentials=$3

curl --silent -u $sr_credentials $sr_url/subjects/$subject/versions/latest | jq -r ".subject" | grep $subject > /dev/null

return $?

URL=$1

EMAIL=$2

PASSWORD=$3

ccloud::validate_expect_installed

echo -e "\n# Login"

OUTPUT=$(

echo "$OUTPUT"

if [[ ! "$OUTPUT" =~ "Logged in as" ]]; then

echo "Failed to log into your cluster. Please check all parameters and run again."

fi

return 0

[ -z "$1" ] && {

echo "ccloud::get_service_account expects one parameter (API Key)"

exit 1

}

[ $# -gt 1 ] && echo "WARN: ccloud::get_service_account function expects one parameter, received two"

local key="$1"

serviceAccount=$(confluent api-key list -o json | jq -r -c 'map(select((.key == "'"$key"'"))) | .[].owner_id')

if [[ "$serviceAccount" == "" ]]; then

echo "ERROR: Could not associate key $key to a service account. Verify your credentials, ensure the API key has a set resource type, and try again."

exit 1

fi

if ! [[ "$serviceAccount" =~ ^sa-[a-z0-9]+$ ]]; then

echo "ERROR: $serviceAccount value is not a valid value for a service account. Verify your credentials, ensure the API key has a set resource type, and try again."

exit 1

fi

echo "$serviceAccount"

return 0

serviceAccount=$1

confluent kafka acl create --allow --service-account $serviceAccount --operations DESCRIBE --cluster-scope

confluent kafka acl create --allow --service-account $serviceAccount --operations CREATE,WRITE --prefix --topic dlq-lcc

confluent kafka acl create --allow --service-account $serviceAccount --operations READ --prefix --consumer-group connect-lcc

return 0

serviceAccount=$1

echo "Confluent Control Center: creating _confluent-command and ACLs for service account $serviceAccount"

confluent kafka topic create _confluent-command --partitions 1

confluent kafka acl create --allow --service-account $serviceAccount --operations WRITE,READ,CREATE --topic _confluent --prefix

confluent kafka acl create --allow --service-account $serviceAccount --operations READ,CREATE --consumer-group _confluent --prefix

return 0

serviceAccount=$1

topic=$2

confluent kafka acl create --allow --service-account $serviceAccount --operations CREATE,WRITE,READ,DESCRIBE,DESCRIBE-CONFIGS,ALTER-CONFIGS --topic $topic

confluent kafka acl create --allow --service-account $serviceAccount --operations DESCRIBE --cluster-scope

return 0

serviceAccount=$1

echo "Connect: creating topics and ACLs for service account $serviceAccount"

TOPIC=connect-demo-configs

confluent kafka topic create $TOPIC --partitions 1 --config "cleanup.policy=compact"

confluent kafka acl create --allow --service-account $serviceAccount --operations WRITE,READ --topic $TOPIC --prefix

TOPIC=connect-demo-offsets

confluent kafka topic create $TOPIC --partitions 6 --config "cleanup.policy=compact"

confluent kafka acl create --allow --service-account $serviceAccount --operations WRITE,READ --topic $TOPIC --prefix

TOPIC=connect-demo-statuses

confluent kafka topic create $TOPIC --partitions 3 --config "cleanup.policy=compact"

confluent kafka acl create --allow --service-account $serviceAccount --operations WRITE,READ --topic $TOPIC --prefix

for TOPIC in _confluent-monitoring _confluent-command ; do

confluent kafka topic create $TOPIC &>/dev/null

confluent kafka acl create --allow --service-account $serviceAccount --operations WRITE,READ --topic $TOPIC --prefix

done

confluent kafka acl create --allow --service-account $serviceAccount --operations READ --consumer-group connect-cloud

echo "Connectors: creating topics and ACLs for service account $serviceAccount"

confluent kafka acl create --allow --service-account $serviceAccount --operations READ --consumer-group connect-replicator

confluent kafka acl create --allow --service-account $serviceAccount --operations describe --cluster-scope

return 0

CLOUD_KEY=$1

CONFIG_FILE=$2

enable_ksqldb=$3

if [ -z "$enable_ksqldb" ]; then

enable_ksqldb=true

fi

ccloud::validate_environment_set || exit 1

ccloud::set_kafka_cluster_use_from_api_key "$CLOUD_KEY" || exit 1

ccloud::validate_schema_registry_up "$SCHEMA_REGISTRY_BASIC_AUTH_USER_INFO" "$SCHEMA_REGISTRY_URL" || exit 1

if $enable_ksqldb ; then

ccloud::validate_ksqldb_up "$KSQLDB_ENDPOINT" || exit 1

ccloud::validate_credentials_ksqldb "$KSQLDB_ENDPOINT" "$CONFIG_FILE" "$KSQLDB_BASIC_AUTH_USER_INFO" || exit 1

fi

confluent environment list | grep '*' &>/dev/null || {

echo "ERROR: could not determine if environment is set. Run 'confluent environment list' and set 'confluent environment use' and try again"

exit 1

}

return 0

[ -z "$1" ] && {

echo "ccloud::set_kafka_cluster_use_from_api_key expects one parameter (API Key)"

exit 1

}

[ $# -gt 1 ] && echo "WARN: ccloud::set_kafka_cluster_use_from_api_key function expects one parameter, received two"

local key="$1"

local kafkaCluster=$(confluent api-key list -o json | jq -r -c 'map(select((.key == "'"$key"'" and .resource_type == "kafka"))) | .[].resource_id')

if [[ "$kafkaCluster" == "" ]]; then

echo "ERROR: Could not associate key $key to a Confluent Cloud Kafka cluster. Verify your credentials, ensure the API key has a set resource type, and try again."

exit 1

fi

confluent kafka cluster use $kafkaCluster

local endpoint=$(confluent kafka cluster describe $kafkaCluster -o json | jq -r ".endpoint" | cut -c 12-)

echo -e "\nAssociated key $key to Confluent Cloud Kafka cluster $kafkaCluster at $endpoint"

return 0

ENVIRONMENT_NAME=$1

if [[ -z "$ENVIRONMENT" ]];

then

ENVIRONMENT=$(ccloud::create_and_use_environment $ENVIRONMENT_NAME)

(($? != 0)) && { echo "$ENVIRONMENT"; exit 1; }

else

confluent environment use $ENVIRONMENT || exit 1

fi

echo "$ENVIRONMENT"

return 0

ccloud::validate_version_cli $CLI_MIN_VERSION || exit 1

QUIET="${QUIET:-false}"

REPLICATION_FACTOR=${REPLICATION_FACTOR:-3}

enable_ksqldb=${1:-false}

EXAMPLE=${EXAMPLE:-ccloud-stack-function}

CHECK_CREDIT_CARD="${CHECK_CREDIT_CARD:-false}"

# Check if credit card is on file, which is required for cluster creation

if $CHECK_CREDIT_CARD && [[ $(confluent admin payment describe) =~ "not found" ]]; then

echo "ERROR: No credit card on file. Add a payment method and try again."

echo "If you are using a cloud provider's Marketplace, see documentation for a workaround: https://docs.confluent.io/platform/current/tutorials/examples/ccloud/docs/ccloud-stack.html#running-with-marketplace"

exit 1

fi

if [[ -z "$SERVICE_ACCOUNT_ID" ]]; then

# Service Account is not received so it will be created

local RANDOM_NUM=$((1 + RANDOM % 1000000))

SERVICE_NAME=${SERVICE_NAME:-"demo-app-$RANDOM_NUM"}

SERVICE_ACCOUNT_ID=$(ccloud::create_service_account $SERVICE_NAME)

fi

if [[ "$SERVICE_NAME" == "" ]]; then

echo "ERROR: SERVICE_NAME is not defined. If you are providing the SERVICE_ACCOUNT_ID to this function please also provide the SERVICE_NAME"

exit 1

fi

echo "Creating Confluent Cloud stack for service account $SERVICE_NAME, ID: $SERVICE_ACCOUNT_ID."

ENVIRONMENT_NAME=${ENVIRONMENT_NAME:-"ccloud-stack-$SERVICE_ACCOUNT_ID-$EXAMPLE"}

ENVIRONMENT=$(ccloud::maybe_create_and_use_environment "$ENVIRONMENT_NAME")

CLUSTER_NAME=${CLUSTER_NAME:-"demo-kafka-cluster-$SERVICE_ACCOUNT_ID"}

CLUSTER_CLOUD="${CLUSTER_CLOUD:-aws}"

CLUSTER_REGION="${CLUSTER_REGION:-us-west-2}"

CLUSTER=$(ccloud::maybe_create_and_use_cluster "$CLUSTER_NAME" $CLUSTER_CLOUD $CLUSTER_REGION)

(($? != 0)) && { echo "$CLUSTER"; exit 1; }

if [[ "$CLUSTER" == "" ]] ; then

echo "Kafka cluster id is empty"

echo "ERROR: Could not create cluster. Please troubleshoot."

exit 1

fi

# Sometimes bootstrap.servers is empty so testing a sleep

sleep 3

BOOTSTRAP_SERVERS=$(confluent kafka cluster describe $CLUSTER -o json | jq -r ".endpoint" | cut -c 12-)

CLUSTER_CREDS=$(ccloud::maybe_create_credentials_resource $SERVICE_ACCOUNT_ID $CLUSTER)

if [[ "$CLUSTER_CREDS" == "" ]] ; then

echo "Credentials are empty"

echo "ERROR: Could not create credentials."

exit 1

fi

MAX_WAIT=720

echo ""

echo "Waiting up to $MAX_WAIT seconds for Confluent Cloud cluster to be ready and for credentials to propagate"

ccloud::retry $MAX_WAIT ccloud::validate_ccloud_cluster_ready || exit 1

# Estimating another 80s wait still sometimes required

WARMUP_TIME=${WARMUP_TIME:-80}

echo -e "Sleeping an additional ${WARMUP_TIME} seconds to ensure propagation of all metadata\n"

sleep $WARMUP_TIME

ccloud::grant_envadmin_access $SERVICE_ACCOUNT_ID $ENVIRONMENT

SCHEMA_REGISTRY_GEO="${SCHEMA_REGISTRY_GEO:-us}"

SCHEMA_REGISTRY=$(ccloud::enable_schema_registry $CLUSTER_CLOUD $SCHEMA_REGISTRY_GEO)

SCHEMA_REGISTRY_ENDPOINT=$(confluent schema-registry cluster describe -o json | jq -r ".endpoint_url")

SCHEMA_REGISTRY_CREDS=$(ccloud::maybe_create_credentials_resource $SERVICE_ACCOUNT_ID $SCHEMA_REGISTRY)

if $enable_ksqldb ; then

KSQLDB_NAME=${KSQLDB_NAME:-"demo-ksqldb-$SERVICE_ACCOUNT_ID"}

KSQLDB=$(ccloud::maybe_create_ksqldb_app "$KSQLDB_NAME" $CLUSTER "$SERVICE_ACCOUNT_ID")

KSQLDB_ENDPOINT=$(confluent ksql cluster describe $KSQLDB -o json | jq -r ".endpoint")

KSQLDB_CREDS=$(ccloud::maybe_create_credentials_resource $SERVICE_ACCOUNT_ID $KSQLDB)

fi

CLOUD_API_KEY=`echo $CLUSTER_CREDS | awk -F: '{print $1}'`

CLOUD_API_SECRET=`echo $CLUSTER_CREDS | awk -F: '{print $2}'`

confluent api-key use $CLOUD_API_KEY --resource ${CLUSTER}

if [[ -z "$SKIP_CONFIG_FILE_WRITE" ]]; then

if [[ -z "$CONFIG_FILE" ]]; then

mkdir -p stack-configs

CONFIG_FILE="stack-configs/java-service-account-$SERVICE_ACCOUNT_ID.config"

fi

cat <<EOF > $CONFIG_FILE