kcl github workflow

Author: skyero-aws

.github/scripts/clean_up_stream_table.sh

@@ -0,0 +1,19 @@
+#!/bin/bash
+
+aws kinesis delete-stream --stream-name $STREAM_NAME || true
+
+# Delete all tables
+for i in {1..10}; do
+  echo "Deleting table $APP_NAME"
+  aws dynamodb delete-table --table-name $APP_NAME && break ||
+  echo "Table deletion failed, attempt $i/10. Retrying DynamoDB Table deletion in $((i * 3)) seconds" && sleep $((i * 3))
+done
+for SUFFIX in "-CoordinatorState" "-WorkerMetricStats"; do
+  if aws dynamodb describe-table --table-name $APP_NAME$SUFFIX &>/dev/null; then
+    echo "Deleting table $APP_NAME$SUFFIX"
+    for i in {1..10}; do
+      aws dynamodb delete-table --table-name $APP_NAME$SUFFIX && break ||
+      echo "Table deletion failed, attempt $i/10. Retrying DynamoDB Table deletion in $((i * 3))s" && sleep $((i * 3))
+    done
+  fi
+done

.github/scripts/create_stream.sh

@@ -0,0 +1,12 @@
+#!/bin/bash
+set -e
+
+for i in {1..10}; do
+  if aws kinesis create-stream --stream-name $STREAM_NAME --shard-count 1; then
+    break
+  else
+    echo "Stream creation failed, attempt $i/10. Waiting $((i * 3)) seconds..."
+    sleep $((i * 3))
+  fi
+done
+aws kinesis wait stream-exists --stream-name $STREAM_NAME

.github/scripts/manipulate_properties.sh

@@ -0,0 +1,29 @@
+#!/bin/bash
+set -e
+
+# Manipulate sample.properties file that the KCL application pulls properties from (ex: streamName, applicationName)
+# Depending on the OS, different properties need to be changed
+if [[ "$RUNNER_OS" == "macOS" ]]; then
+  sed -i "" "s/kclpysample/$STREAM_NAME/g" samples/sample.properties
+  sed -i "" "s/PythonKCLSample/$APP_NAME/g" samples/sample.properties
+  sed -i "" 's/us-east-5/us-east-1/g' samples/sample.properties
+  grep -v "idleTimeBetweenReadsInMillis" samples/sample.properties > samples/temp.properties
+  echo "idleTimeBetweenReadsInMillis = 250" >> samples/temp.properties
+  mv samples/temp.properties samples/sample.properties
+elif [[ "$RUNNER_OS" == "Linux" || "$RUNNER_OS" == "Windows" ]]; then
+  sed -i "s/kclpysample/$STREAM_NAME/g" samples/sample.properties
+  sed -i "s/PythonKCLSample/$APP_NAME/g" samples/sample.properties
+  sed -i 's/us-east-5/us-east-1/g' samples/sample.properties
+  sed -i "/idleTimeBetweenReadsInMillis/c\idleTimeBetweenReadsInMillis = 250" samples/sample.properties
+
+  if [[ "$RUNNER_OS" == "Windows" ]]; then
+    echo '@echo off' > samples/run_script.bat
+    echo 'python %~dp0\sample_kclpy_app.py %*' >> samples/run_script.bat
+    sed -i 's/executableName = sample_kclpy_app.py/executableName = samples\/run_script.bat/' samples/sample.properties
+  fi
+else
+  echo "Unknown OS: $RUNNER_OS"
+  exit 1
+fi
+
+cat samples/sample.properties

.github/scripts/put_words_to_stream.sh

@@ -0,0 +1,15 @@
+#!/bin/bash
+set -e
+
+sample_kinesis_wordputter.py --stream $STREAM_NAME -w cat -w dog -w bird -w lobster -w octopus
+
+# Get records from stream to verify they exist before continuing
+SHARD_ITERATOR=$(aws kinesis get-shard-iterator --stream-name $STREAM_NAME --shard-id shardId-000000000000 --shard-iterator-type TRIM_HORIZON --query 'ShardIterator' --output text)
+INITIAL_RECORDS=$(aws kinesis get-records --shard-iterator $SHARD_ITERATOR)
+RECORD_COUNT_BEFORE=$(echo $INITIAL_RECORDS | jq '.Records | length')
+
+if [ "$RECORD_COUNT_BEFORE" -eq 0 ]; then
+  echo "No records found in stream. Test cannot proceed."
+  exit 1
+fi
+echo "Found $RECORD_COUNT_BEFORE records in stream before KCL start"

.github/scripts/start_kcl.sh

@@ -0,0 +1,28 @@
+#!/bin/bash
+set -e
+set -o pipefail
+
+chmod +x samples/sample.properties
+chmod +x samples/sample_kclpy_app.py
+
+# Get records from stream to verify they exist before continuing
+SHARD_ITERATOR=$(aws kinesis get-shard-iterator --stream-name $STREAM_NAME --shard-id shardId-000000000000 --shard-iterator-type TRIM_HORIZON --query 'ShardIterator' --output text)
+INITIAL_RECORDS=$(aws kinesis get-records --shard-iterator $SHARD_ITERATOR)
+RECORD_COUNT_BEFORE=$(echo $INITIAL_RECORDS | jq '.Records | length')
+
+echo "Found $RECORD_COUNT_BEFORE records in stream before KCL start"
+
+if [[ "$RUNNER_OS" == "macOS" ]]; then
+  brew install coreutils
+  KCL_COMMAND=$(amazon_kclpy_helper.py --print_command --java $(which java) --properties samples/sample.properties)
+  gtimeout $RUN_TIME_SECONDS $KCL_COMMAND 2>&1 | tee kcl_output.log  || [ $? -eq 124 ]
+elif [[ "$RUNNER_OS" == "Linux" || "$RUNNER_OS" == "Windows" ]]; then
+  KCL_COMMAND=$(amazon_kclpy_helper.py --print_command --java $(which java) --properties samples/sample.properties)
+  timeout $RUN_TIME_SECONDS $KCL_COMMAND 2>&1 | tee kcl_output.log || [ $? -eq 124 ]
+else
+  echo "Unknown OS: $RUNNER_OS"
+  exit 1
+fi
+
+echo "==========ERROR LOGS=========="
+grep -i error kcl_output.log || echo "No errors found in logs"

.github/scripts/verify_kcl.sh

@@ -0,0 +1,20 @@
+#!/bin/bash
+set -e
+
+NUM_LEASES_FOUND=$(aws dynamodb scan --table-name $APP_NAME --select "COUNT" --query "Count" --output text || echo "0")
+NUM_CHECKPOINTS_FOUND=$(aws dynamodb scan --table-name $APP_NAME --select "COUNT" --filter-expression "attribute_exists(checkpoint) AND checkpoint <> :trim_horizon" --expression-attribute-values '{":trim_horizon": {"S": "TRIM_HORIZON"}}' --query "Count" --output text || echo "0")
+
+echo "Found $NUM_LEASES_FOUND leases and $NUM_CHECKPOINTS_FOUND non-TRIM-HORIZON checkpoint in DynamoDB"
+
+echo "Printing checkpoint values"
+aws dynamodb scan --table-name $APP_NAME --projection-expression "leaseKey,checkpoint" --output json
+
+if [ "$NUM_LEASES_FOUND" -gt 0 ] && [ "$NUM_CHECKPOINTS_FOUND" -gt 0 ]; then
+  echo "Test passed: Found both leases and non-TRIM_HORIZON checkpoints in DDB (KCL is fully functional)"
+  exit 0
+else
+  echo "Test failed: KCL not fully functional"
+  echo "Lease(s) found: $NUM_LEASES_FOUND"
+  echo "non-TRIM_HORIZON checkpoint(s) found: $NUM_CHECKPOINTS_FOUND"
+  exit 1
+fi