diff --git a/.ci/examples/connect/pubsub.nar b/.ci/examples/connect/pubsub.nar
new file mode 100644
index 000000000..4422f24db
Binary files /dev/null and b/.ci/examples/connect/pubsub.nar differ
diff --git a/.ci/helm.sh b/.ci/helm.sh
index bab77af7f..357e6c50b 100644
--- a/.ci/helm.sh
+++ b/.ci/helm.sh
@@ -608,4 +608,15 @@ function ci::verify_env() {
     return 0
   fi
   return 1
+}
+
+function ci::verify_pod_log() {
+  pod=$1
+  log=$2
+  sleep 30
+  kubectl logs --tail=-1 $pod | grep $log
+  while [[ $? -ne 0 ]]; do
+      sleep 5
+      kubectl logs --tail=-1 $pod | grep $log
+  done
 }
\ No newline at end of file
diff --git a/.ci/tests/integration-oauth2/cases/google-pubsub-source/manifests.yaml b/.ci/tests/integration-oauth2/cases/google-pubsub-source/manifests.yaml
new file mode 100644
index 000000000..efca200bf
--- /dev/null
+++ b/.ci/tests/integration-oauth2/cases/google-pubsub-source/manifests.yaml
@@ -0,0 +1,62 @@
+apiVersion: compute.functionmesh.io/v1alpha1
+kind: Source
+metadata:
+  name: pubsub-source
+  namespace: default
+spec:
+  className: org.apache.pulsar.ecosystem.io.pubsub.PubsubSource
+  clusterName: test-pulsar
+  forwardSourceMessageProperty: true
+  image: streamnative/pulsar-functions-pulsarctl-java-runner:3.2.2.1
+  java:
+    jar: /tmp/pulsar-io-google-pubsub.nar
+    jarLocation: function://public/default/pubsub
+  minReplicas: 1
+  name: test
+  namespace: default
+  output:
+    producerConf: {}
+    topic: persistent://public/default/test-pubsub-output
+    typeClassName: '[B'
+  processingGuarantee: atleast_once
+  pulsar:
+    pulsarConfig: test-pulsar
+    authConfig:
+      oauth2Config:
+        audience: urn:sn:pulsar:sndev:test
+        issuerUrl: https://auth.sncloud-stg.dev/
+        keySecretName: sn-platform-oauth2-private-key
+        keySecretKey: auth.json
+  replicas: 1
+  resources:
+    limits:
+      cpu: "0.2"
+      memory: 1.1G
+    requests:
+      cpu: "0.1"
+      memory: 1G
+  secretsMap:
+    pubsubCredential:
+      key: GCP_KEY
+      path: gcp-account
+  sourceConfig:
+    pubsubProjectId: sncloud-dev-pengcheng
+    pubsubTopicId: test-input
+  tenant: public
+---
+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
+---
+apiVersion: v1
+data:
+  GCP_KEY: GCP_KEY_DATA
+kind: Secret
+metadata:
+  name: gcp-account
+  namespace: default
+type: Opaque
\ No newline at end of file
diff --git a/.ci/tests/integration-oauth2/cases/google-pubsub-source/verify.sh b/.ci/tests/integration-oauth2/cases/google-pubsub-source/verify.sh
new file mode 100644
index 000000000..d3108b06c
--- /dev/null
+++ b/.ci/tests/integration-oauth2/cases/google-pubsub-source/verify.sh
@@ -0,0 +1,51 @@
+#!/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
+
+kubectl apply -f "${BASE_DIR}"/.ci/tests/integration-oauth2/cases/google-pubsub-source/manifests.yaml > /dev/null 2>&1
+
+verify_fm_result=$(ci::verify_function_mesh pubsub-source 2>&1)
+if [ $? -ne 0 ]; then
+  echo "$verify_fm_result"
+  kubectl delete -f "${BASE_DIR}"/.ci/tests/integration-oauth2/cases/google-pubsub-source/manifests.yaml > /dev/null 2>&1 || true
+  exit 1
+fi
+
+verify_pod_log_result=$(NAMESPACE=${PULSAR_NAMESPACE} ci::verify_pod_log pubsub-source-source-0 "listening for messages on projects/sncloud-dev-pengcheng/subscriptions/test-input" 2>&1)
+if [ $? -ne 0 ]; then
+  echo "$verify_pod_log_result"
+  kubectl delete -f "${BASE_DIR}"/.ci/tests/integration-oauth2/cases/google-pubsub-source/manifests.yaml > /dev/null 2>&1 || true
+  exit 1
+fi
+
+kubectl delete -f "${BASE_DIR}"/.ci/tests/integration-oauth2/cases/google-pubsub-source/manifests.yaml > /dev/null 2>&1 || true
diff --git a/.ci/tests/integration-oauth2/e2e.yaml b/.ci/tests/integration-oauth2/e2e.yaml
index 237d24932..5bb5e9f7a 100644
--- a/.ci/tests/integration-oauth2/e2e.yaml
+++ b/.ci/tests/integration-oauth2/e2e.yaml
@@ -73,6 +73,7 @@ setup:
       command: |
         bash .ci/upload_function_with_oauth.sh java
         bash .ci/upload_function_with_oauth.sh py
+        bash .ci/upload_function_with_oauth.sh pubsub
 
     # testing download packages from http
     - name: start nginx http server
@@ -131,3 +132,5 @@ verify:
       expected: expected.data.yaml
     - query: timeout 5m bash .ci/tests/integration-oauth2/cases/batch-source/verify.sh
       expected: expected.data.yaml
+    - query: timeout 5m bash .ci/tests/integration-oauth2/cases/google-pubsub-source/verify.sh
+      expected: expected.data.yaml
diff --git a/.ci/upload_function_with_oauth.sh b/.ci/upload_function_with_oauth.sh
index 2ab6f72f9..f6867478a 100755
--- a/.ci/upload_function_with_oauth.sh
+++ b/.ci/upload_function_with_oauth.sh
@@ -44,4 +44,8 @@ case ${1} in
     kubectl cp "${PULSAR_HOME}/.ci/examples/go-examples" "${NAMESPACE}/${CLUSTER}-pulsar-broker-0:/pulsar/"
     kubectl exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- sh -c 'bin/pulsar-admin --auth-plugin $brokerClientAuthenticationPlugin --auth-params $brokerClientAuthenticationParameters packages upload function://public/default/test-go-function --path /pulsar/go-examples/exclamationFunc --description "test golang function"'
     ;;
+  pubsub)
+    kubectl cp "${PULSAR_HOME}/.ci/examples/connect" "${NAMESPACE}/${CLUSTER}-pulsar-broker-0:/pulsar/connect"
+    kubectl exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- sh -c 'bin/pulsar-admin --auth-plugin $brokerClientAuthenticationPlugin --auth-params $brokerClientAuthenticationParameters packages upload function://public/default/pubsub --path /pulsar/connect/pubsub.nar --description "pub sub io"'
+    ;;
 esac