fix logging window function

.ci/helm.sh

@@ -96,7 +96,7 @@ function ci::install_pulsar_charts() {
     echo "wait until pulsar init job is completed"
     succeeded_num=0
     while [[ ${succeeded_num} -lt 1 ]]; do
-        sleep 10s
+        sleep 10
         kubectl get pods -n ${NAMESPACE}
         succeeded_num=$(kubectl get jobs -n ${NAMESPACE} sn-platform-pulsar-pulsar-init -o jsonpath='{.status.succeeded}')
     done
@@ -125,7 +125,7 @@ function ci::verify_function_mesh() {
 
     num=0
     while [[ ${num} -lt 1 ]]; do
-        sleep 5s
+        sleep 5
         kubectl get pods
         num=$(kubectl get pods -l compute.functionmesh.io/name="${FUNCTION_NAME}" | wc -l)
     done
@@ -134,7 +134,7 @@ function ci::verify_function_mesh() {
 
     num=0
     while [[ ${num} -lt 1 ]]; do
-        sleep 5s
+        sleep 5
         kubectl get pods -l compute.functionmesh.io/name="${FUNCTION_NAME}"
         kubectl logs -l compute.functionmesh.io/name="${FUNCTION_NAME}" --all-containers=true --tail=50 || true
         num=$(kubectl logs -l compute.functionmesh.io/name="${FUNCTION_NAME}" --all-containers=true --tail=-1 | grep "Created producer\|Created consumer\|Subscribed to topic" | wc -l)
@@ -164,7 +164,7 @@ function ci::test_function_runners() {
     kubectl exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- bin/pulsar-admin functions create --tenant public --namespace default --name test-java --className org.apache.pulsar.functions.api.examples.ExclamationFunction --inputs persistent://public/default/test-java-input --jar /pulsar/examples/api-examples.jar --cpu 0.1
     function_num=0
     while [[ ${function_num} -lt 1 ]]; do
-        sleep 5s
+        sleep 5
         kubectl get pods -n ${NAMESPACE}
         function_num=$(kubectl get pods -n ${NAMESPACE} -l name=test-java --no-headers | wc -l)
     done
@@ -178,7 +178,7 @@ function ci::test_function_runners() {
     kubectl exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- bin/pulsar-admin functions create --tenant public --namespace default --name test-python --classname exclamation_function.ExclamationFunction --inputs persistent://public/default/test-python-input --py /pulsar/examples/python-examples/exclamation_function.py --cpu 0.1
     function_num=0
     while [[ ${function_num} -lt 1 ]]; do
-        sleep 5s
+        sleep 5
         kubectl get pods -n ${NAMESPACE}
         function_num=$(kubectl get pods -n ${NAMESPACE} -l name=test-python --no-headers | wc -l)
     done
@@ -194,7 +194,7 @@ function ci::test_function_runners() {
     kubectl exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- bin/pulsar-admin functions create --tenant public --namespace default --name test-go --inputs persistent://public/default/test-go-input --go /pulsar/go-examples/exclamationFunc --cpu 0.1
     function_num=0
     while [[ ${function_num} -lt 1 ]]; do
-        sleep 5s
+        sleep 5
         kubectl get pods -n ${NAMESPACE}
         function_num=$(kubectl get pods -n ${NAMESPACE} -l name=test-go --no-headers | wc -l)
     done
@@ -273,6 +273,13 @@ function ci::verify_crypto_function() {
     ci::verify_function_with_encryption "persistent://public/default/java-function-crypto-input-topic" "persistent://public/default/java-function-crypto-output-topic" "test-message" "test-message!" 10
 }
 
+function ci::send_test_data() {
+    inputtopic=$1
+    inputmessage=$2
+    kubectl exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- bin/pulsar-client produce -m "${inputmessage}" -n 100 "${inputtopic}"
+    return 0
+}
+
 function ci::verify_exclamation_function() {
     inputtopic=$1
     outputtopic=$2
@@ -330,7 +337,7 @@ function ci::verify_functionmesh_reconciliation() {
     function_name="${MESH_NAME}"-function-"${POSTFIX}"
     find=false
     while ! ${find}; do
-        sleep 5s
+        sleep 5
         num=$(kubectl get functions.compute.functionmesh.io --no-headers | wc -l)
         if [[ ${num} -ne 1 ]]; then
             continue
@@ -344,7 +351,7 @@ function ci::verify_functionmesh_reconciliation() {
     sink_name="${MESH_NAME}"-sink-"${POSTFIX}"
     find=false
     while ! ${find}; do
-        sleep 5s
+        sleep 5
         num=$(kubectl get sinks.compute.functionmesh.io --no-headers | wc -l)
         if [[ ${num} -ne 1 ]]; then
             continue
@@ -358,7 +365,7 @@ function ci::verify_functionmesh_reconciliation() {
     source_name="${MESH_NAME}"-source-"${POSTFIX}"
     find=false
     while ! ${find}; do
-        sleep 5s
+        sleep 5
         num=$(kubectl get sources.compute.functionmesh.io --no-headers | wc -l)
         if [[ ${num} -ne 1 ]]; then
             continue

.ci/tests/integration-oauth2/e2e.yaml

@@ -45,7 +45,7 @@ setup:
         echo "wait until pulsar init job is completed"
         succeeded_num=0
         while [[ ${succeeded_num} -lt 1 ]]; do
-          sleep 10s
+          sleep 10
           kubectl get pods -n ${PULSAR_NAMESPACE}
           succeeded_num=$(kubectl get jobs -n ${PULSAR_NAMESPACE} ${PULSAR_RELEASE_NAME}-pulsar-pulsar-init -o jsonpath='{.status.succeeded}')
         done

.ci/tests/integration/cases/elasticsearch-sink/verify.sh

@@ -41,7 +41,7 @@ function install_elasticsearch_cluster() {
     kubectl apply -f https://download.elastic.co/downloads/eck/2.3.0/operator.yaml
     num=0
     while [[ ${num} -lt 1 ]]; do
-        sleep 5s
+        sleep 5
         kubectl get pods -n elastic-system
         num=$(kubectl get pods -n elastic-system -l control-plane=elastic-operator | wc -l)
     done
@@ -51,7 +51,7 @@ function install_elasticsearch_cluster() {
     kubectl apply -f "${es_file}"
     num=0
     while [[ ${num} -lt 1 ]]; do
-        sleep 5s
+        sleep 5
         kubectl get pods
         num=$(kubectl get pods -l elasticsearch.k8s.elastic.co/cluster-name=quickstart | wc -l)
     done
@@ -65,7 +65,7 @@ function uninstall_elasticsearch_cluster() {
         if [ $? -eq 1 ]; then
             break
         fi
-        sleep 5s
+        sleep 5
     done
 
     kubectl delete -f https://download.elastic.co/downloads/eck/2.3.0/operator.yaml

.ci/tests/integration/cases/logging-window-function/manifests.yaml

@@ -0,0 +1,47 @@
+apiVersion: compute.functionmesh.io/v1alpha1
+kind: Function
+metadata:
+  name: window-function-sample
+  namespace: default
+spec:
+  image: streamnative/pulsar-functions-java-sample:2.9.2.23
+  className: org.apache.pulsar.functions.api.examples.window.LoggingWindowFunction
+  forwardSourceMessageProperty: true
+  maxPendingAsyncRequests: 1000
+  replicas: 1
+  logTopic: persistent://public/default/window-function-logs
+  input:
+    topics:
+      - persistent://public/default/window-function-input-topic
+    typeClassName: java.lang.String
+  output:
+    topic: persistent://public/default/window-function-output-topic
+    typeClassName: java.lang.String
+  resources:
+    requests:
+      cpu: 50m
+      memory: 1G
+    limits:
+      cpu: "0.2"
+      memory: 1.1G
+  pulsar:
+    pulsarConfig: "test-pulsar"
+  java:
+    jar: /pulsar/examples/api-examples.jar
+  clusterName: test-pulsar
+  autoAck: true
+  statefulConfig:
+    pulsar:
+      serviceUrl: "bk://sn-platform-pulsar-bookie.default.svc.cluster.local:4181"
+  windowConfig:
+    windowLengthCount: 10
+    slidingIntervalCount: 5
+  subscriptionPosition: earliest
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: test-pulsar
+data:
+  webServiceURL: http://sn-platform-pulsar-broker.default.svc.cluster.local:8080
+  brokerServiceURL: pulsar://sn-platform-pulsar-broker.default.svc.cluster.local:6650

.ci/tests/integration/cases/logging-window-function/verify.sh

@@ -0,0 +1,66 @@
+#!/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.
+#
+
+set -e
+
+E2E_DIR=$(dirname "$0")
+BASE_DIR=$(cd "${E2E_DIR}"/../../../../..;pwd)
+PULSAR_NAMESPACE=${PULSAR_NAMESPACE:-"default"}
+PULSAR_RELEASE_NAME=${PULSAR_RELEASE_NAME:-"sn-platform"}
+E2E_KUBECONFIG=${E2E_KUBECONFIG:-"/tmp/e2e-k8s.config"}
+
+source "${BASE_DIR}"/.ci/helm.sh
+
+if [ ! "$KUBECONFIG" ]; then
+  export KUBECONFIG=${E2E_KUBECONFIG}
+fi
+
+manifests_file="${BASE_DIR}"/.ci/tests/integration/cases/logging-window-function/manifests.yaml
+
+kubectl apply -f "${manifests_file}" > /dev/null 2>&1
+
+verify_fm_result=$(ci::verify_function_mesh window-function-sample 2>&1)
+if [ $? -ne 0 ]; then
+  echo "$verify_fm_result"
+  kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true
+  exit 1
+fi
+
+verify_java_result=$(NAMESPACE=${PULSAR_NAMESPACE} CLUSTER=${PULSAR_RELEASE_NAME} ci::send_test_data "persistent://public/default/window-function-input-topic" "test-message" 2>&1)
+if [ $? -ne 0 ]; then
+  echo "$verify_java_result"
+  kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true
+  exit 1
+fi
+
+verify_log_result=$(kubectl logs -l compute.functionmesh.io/name=window-function-sample --tail=-1 | grep -e "-window-log" | wc -l)
+if [ $verify_log_result -ne 0 ]; then
+  sub_name=$(echo $RANDOM | md5sum | head -c 20; echo;)
+  verify_log_topic_result=$(kubectl exec -n ${PULSAR_NAMESPACE} ${PULSAR_RELEASE_NAME}-pulsar-broker-0 -- bin/pulsar-client consume -n 10 -s $sub_name --subscription-position Earliest "persistent://public/default/window-function-logs" | grep  -e "-window-log" | wc -l)
+  if [ $verify_log_topic_result -ne 0 ]; then
+    echo "e2e-test: ok" | yq eval -
+  else
+    echo "$verify_log_topic_result"
+  fi
+else
+  echo "$verify_log_result"
+fi
+
+kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true

.ci/tests/integration/cases/mongodb-source/verify.sh

@@ -44,7 +44,7 @@ function install_mongodb_server() {
     kubectl apply -f "${mongodb_file}"
     num=0
     while [[ ${num} -lt 3 ]]; do
-        sleep 5s
+        sleep 5
         kubectl get pods
         num=$(kubectl get pods -l role=mongo | wc -l)
     done

.ci/tests/integration/e2e.yaml

@@ -40,7 +40,7 @@ setup:
         echo "wait until pulsar init job is completed"
         succeeded_num=0
         while [[ ${succeeded_num} -lt 1 ]]; do
-          sleep 10s
+          sleep 10
           kubectl get pods -n ${PULSAR_NAMESPACE}
           succeeded_num=$(kubectl get jobs -n ${PULSAR_NAMESPACE} ${PULSAR_RELEASE_NAME}-pulsar-pulsar-init -o jsonpath='{.status.succeeded}')
         done
@@ -137,3 +137,5 @@ verify:
       expected: expected.data.yaml
     - query: timeout 5m bash .ci/tests/integration/cases/python-log-level/verify.sh
       expected: expected.data.yaml
+    - query: timeout 5m bash .ci/tests/integration/cases/logging-window-function/verify.sh
+      expected: expected.data.yaml

.ci/tests/integration/e2e_with_tls.yaml

@@ -40,7 +40,7 @@ setup:
         echo "wait until pulsar init job is completed"
         succeeded_num=0
         while [[ ${succeeded_num} -lt 1 ]]; do
-          sleep 10s
+          sleep 10
           kubectl get pods -n ${PULSAR_NAMESPACE}
           succeeded_num=$(kubectl get jobs -n ${PULSAR_NAMESPACE} ${PULSAR_RELEASE_NAME}-pulsar-pulsar-init -o jsonpath='{.status.succeeded}')
         done

.github/workflows/olm-verify.yml

@@ -77,7 +77,7 @@ jobs:
           kubectl get catalogsource -n olm
           kubectl get catalogsource my-test-catalog -n olm -o yaml
           kubectl get pod -n olm
-          sleep 60s
+          sleep 60
           kubectl -n olm wait --for=condition=ready pod -l olm.catalogSource=my-test-catalog --timeout=600s
           kubectl get catalogsource my-test-catalog -n olm -o yaml
 
@@ -88,7 +88,7 @@ jobs:
       - name: Create a Subscription
         run: |
           kubectl apply -f .ci/olm-tests/subs.yml
-          sleep 60s
+          sleep 60
 
       - name: Verify Operator health
         run: |

images/pulsar-functions-base-runner/Dockerfile

@@ -33,8 +33,8 @@ RUN echo "VERSION_TAG=$(VERSION_TAG)" && \
         echo "Pulsar version is 2.8, use java 1.8" && \
         export JRE_PACKAGE_NAME=openjdk-8-jre-headless; \
     elif [ $VERSION_MAJOR -eq 2 ] && [ $VERSION_MINOR -eq 9 ]; then \
-        echo "Pulsar version is 2.9, use java 1.8" && \
-        export JRE_PACKAGE_NAME=openjdk-8-jre-headless; \
+        echo "Pulsar version is 2.9, use java 11" && \
+        export JRE_PACKAGE_NAME=openjdk-11-jre-headless; \
     elif [ $VERSION_MAJOR -eq 2 ] && [ $VERSION_MINOR -eq 10 ]; then \
         echo "Pulsar version is 2.10, use java 11" && \
         export JRE_PACKAGE_NAME=openjdk-11-jre-headless; \