Merge branch 'improvement/ZENKO-4286' into q/2.11

Author: bert-e

.github/scripts/end2end/common.sh

@@ -61,3 +61,41 @@ wait_for_all_pods_behind_services() {
         done
     done
 }
+
+# wait for consumer group to be in a stable state (no rebance + at least one consumer connected)
+wait_for_consumer_group() {
+    namespace=$1
+    # Getting the name of the first kafka pod
+    kafka_pod=$(kubectl get pods -n $namespace -l brokerId=0,kafka_cr=end2end-base-queue,app=kafka -o jsonpath='{.items[0].metadata.name}')
+    consumer_group=$2
+    # When a pod is restarted the previous consumer is kept in the group until the session timeout expires
+    expected_members=$3
+    timeout_s=$4
+    interval_s=${5:-5}
+    kubectl exec -it $kafka_pod -n $namespace -- bash -c '
+export KAFKA_OPTS=
+consumer_group=$1
+expected_members=$2
+timeout_s=$3
+interval_s=$4
+start_time=$(date +%s)
+while true; do
+    # The state becomes "Stable" when no rebalance is happening and at least one consumer is connected
+    state=$(kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group $consumer_group --state | awk '"'"'NF>1 && $(NF-1) != "STATE" {print (NF>1?$(NF-1):"None")} {next}'"'"')
+    members=$(kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group $consumer_group --state | awk '"'"'NF>1 && $NF != "#MEMBERS" {print (NF>1?$NF:"None")} {next}'"'"')
+    echo "Consumer group $consumer_group state: $state, members: $members"
+    if [ "$state" == "Stable" ] && [ "$members" -eq "$expected_members" ]; then
+        echo "Consumer group $consumer_group is now consuming."
+        exit 0
+    fi
+    # Check if we have reached the timeout
+    current_time=$(date +%s)
+    elapsed_time=$((current_time - start_time))
+    if [ "$elapsed_time" -ge "$timeout_s" ]; then
+        echo "Error: Timed out waiting for consumer group $consumer_group to start consuming."
+        exit 1
+    fi
+    sleep $interval_s
+done
+' -- "$consumer_group" "$expected_members" "$timeout_s" "$interval_s"
+}

.github/scripts/end2end/configs/zenko.yaml

@@ -34,7 +34,7 @@ spec:
     ingestionProcessor:
       concurrency: 2
     logging:
-      logLevel: trace
+      logLevel: debug
   mongodb:
     provider: External
     endpoints:

.github/scripts/end2end/configure-e2e.sh

@@ -109,3 +109,10 @@ sleep 10
 
 kubectl wait --for condition=DeploymentFailure=false --timeout 25m -n ${NAMESPACE} zenko/${ZENKO_NAME}
 kubectl wait --for condition=DeploymentInProgress=false --timeout 25m -n ${NAMESPACE} zenko/${ZENKO_NAME}
+
+
+if [ $ENABLE_RING_TESTS = true ]; then
+  # wait for ingestion processor to start consuming from Kafka
+  ingestion_processor_replicas=$(kubectl -n $NAMESPACE get deploy/end2end-backbeat-ingestion-processor -o jsonpath='{.spec.replicas}')
+  wait_for_consumer_group $NAMESPACE $UUID.backbeat-ingestion-group $ingestion_processor_replicas 300
+fi

tests/zenko_tests/node_tests/backbeat/tests/ingestion/ingestionS3CPauseResume.js

@@ -93,7 +93,6 @@ describe('Ingestion pause resume', function () {
             'non-existent-location',
             (err, data) => {
                 assert.ifError(err);
-                assert.strictEqual(data.code, 404);
                 assert.strictEqual(data.RouteNotFound, true);
                 return done();
             },

tests/zenko_tests/node_tests/package.json

@@ -46,14 +46,14 @@
     "test_expiration": "mocha --tags ${MOCHA_TAGS} --exit -t 900000 --reporter mocha-multi-reporters --reporter-options configFile=config.json backbeat/tests/lifecycle/expiration.js",
     "test_transition": "mocha --tags ${MOCHA_TAGS} --exit -t 900000 --reporter mocha-multi-reporters --reporter-options configFile=config.json backbeat/tests/lifecycle/transition.js",
     "test_lifecycle": "mocha --tags ${MOCHA_TAGS} --exit -t 1800000 --reporter mocha-multi-reporters --reporter-options configFile=config.json --recursive backbeat/tests/lifecycle",
-    "test_ingestion_oob_s3c": "mocha --tags ${MOCHA_TAGS} --exit -t 100000 --reporter mocha-multi-reporters --reporter-options configFile=config.json --recursive backbeat/tests/ingestion",
+    "test_ingestion_oob_s3c": "mocha --tags ${MOCHA_TAGS} --exit -t 180000 --reporter mocha-multi-reporters --reporter-options configFile=config.json --recursive backbeat/tests/ingestion",
     "test_location_quota": "mocha --tags ${MOCHA_TAGS} --exit -t 10000 --reporter mocha-multi-reporters --reporter-options configFile=config.json  --recursive cloudserver/locationQuota/tests",
     "test_bucket_get_v2": "mocha --tags ${MOCHA_TAGS} --exit -t 10000 --reporter mocha-multi-reporters --reporter-options configFile=config.json --recursive cloudserver/bucketGetV2/tests",
     "test_bucket_policy": "mocha --tags ${MOCHA_TAGS} --exit -t 10000 --reporter mocha-multi-reporters --reporter-options configFile=config.json --recursive cloudserver/bucketPolicy/tests",
     "test_operator": "mocha --tags ${MOCHA_TAGS} --exit -t 10000 --reporter mocha-multi-reporters --reporter-options configFile=config.json ./init_test.js",
     "test_smoke": "mocha --tags ${MOCHA_TAGS} --exit -t 10000 --reporter mocha-multi-reporters --reporter-options configFile=config.json --recursive smoke_tests",
     "test_iam_policies": "mocha --tags ${MOCHA_TAGS} --exit -t 15000 --reporter mocha-multi-reporters --reporter-options configFile=config.json --recursive iam_policies",
-    "test_all_extensions": "run-p --aggregate-output test_aws_crr test_expiration test_transition",
+    "test_all_extensions": "run-p --aggregate-output test_aws_crr test_expiration test_transition test_ingestion_oob_s3c",
     "test_object_api": "mocha --tags ${MOCHA_TAGS} --exit -t 10000 --reporter mocha-multi-reporters --reporter-options configFile=config.json --recursive cloudserver/keyFormatVersion/tests",
     "lint": "eslint $(find . -name '*.js' -not -path '*/node_modules/*')"
   },