Merge remote-tracking branch 'up/master' into fix-python-fn-nack-delay

# Conflicts:
#	pulsar-functions/instance/src/test/python/test_python_instance.py

Author: david-streamlio

.github/workflows/ci-python-functions.yaml

@@ -0,0 +1,90 @@
+#
+# 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.
+#
+
+name: CI - Python Functions
+on:
+  pull_request:
+    branches:
+      - master
+    paths:
+      - '.github/workflows/**'
+      - 'pulsar-functions/instance/src/main/python/**'
+      - 'pulsar-functions/instance/src/test/python/**'
+      - 'pulsar-functions/instance/src/scripts/**'
+  workflow_dispatch:
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  preconditions:
+    name: Preconditions
+    runs-on: ubuntu-24.04
+    outputs:
+      docs_only: ${{ steps.check_changes.outputs.docs_only }}
+    steps:
+      - name: checkout
+        uses: actions/checkout@v6
+
+      - name: Detect changed files
+        id: changes
+        uses: apache/pulsar-test-infra/paths-filter@master
+        with:
+          filters: .github/changes-filter.yaml
+          list-files: csv
+
+      - name: Check changed files
+        id: check_changes
+        run: |
+          if [[ "${GITHUB_EVENT_NAME}" != "schedule" ]]; then
+            echo "docs_only=${{ fromJSON(steps.changes.outputs.all_count) == fromJSON(steps.changes.outputs.docs_count) && fromJSON(steps.changes.outputs.docs_count) > 0 }}" >> $GITHUB_OUTPUT
+          else
+            echo docs_only=false >> $GITHUB_OUTPUT
+          fi
+
+      - name: Check if the PR is ready for running CI
+        if: ${{ steps.check_changes.outputs.docs_only != 'true' && github.repository == 'apache/pulsar' && github.event_name == 'pull_request' }}
+        uses: ./.github/actions/check-pr-ready-to-test
+
+  instance-tests:
+    needs: preconditions
+    if: ${{ needs.preconditions.outputs.docs_only != 'true' }}
+    name: Python ${{ matrix.python-version }} Functions instance tests
+    runs-on: ubuntu-24.04
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: ['3.12', '3.13']
+
+    steps:
+      - name: checkout
+        uses: actions/checkout@v6
+
+      - name: Tune Runner VM
+        uses: ./.github/actions/tune-runner-vm
+
+      - name: Set up Python
+        uses: actions/setup-python@v6
+        with:
+          python-version: ${{ matrix.python-version }}
+
+      - name: Run Python instance tests
+        run: |
+          ./pulsar-functions/instance/src/scripts/run_python_instance_tests.sh

conf/broker.conf

@@ -2021,6 +2021,12 @@ exposeBundlesMetricsInPrometheus=false
 # Default is false.
 exposeCustomTopicMetricLabelsEnabled=false
 
+# Enable computing the age of the oldest unacknowledged message for each subscription and exposing it
+# through topic stats and Prometheus.
+# When disabled, the broker skips computing per-subscription backlog age and the admin API field
+# SubscriptionStats.oldestBacklogMessageAgeSeconds remains -1. Default is false.
+exposeSubscriptionBacklogAgeInPrometheus=false
+
 # A comma-separated list of Topic Property keys that are allowed to be exposed as metrics.
 # Only these keys can be set as custom metric labels on topics.
 # Example: sla_tier,data_sensitivity,cost_center,app_owner

conf/standalone.conf

@@ -1290,6 +1290,12 @@ exposePublisherStats=true
 # Default is false.
 exposePreciseBacklogInPrometheus=false
 
+# Enable computing the age of the oldest unacknowledged message for each subscription and exposing it
+# through topic stats and Prometheus.
+# When disabled, the broker skips computing per-subscription backlog age and the admin API field
+# SubscriptionStats.oldestBacklogMessageAgeSeconds remains -1. Default is false.
+exposeSubscriptionBacklogAgeInPrometheus=false
+
 # Enable splitting topic and partition label in Prometheus.
 # If enabled, a topic name will split into 2 parts, one is topic name without partition index,
 # another one is partition index, e.g. (topic=xxx, partition=0).

pip/pip-382.md

@@ -4,7 +4,7 @@
 
 Pulsar has a metric that indicates load topic failed: `topic_load_failed_total`, it will be increased at the following cases
 - The target bundle in unloading.
-- Failed to load policies.
+- Failed to load namespace-level or topic-level policies.
 - Failed to load up Managed Ledger.
 - Failed to read Metadata store.
 - Topic initialize fails, such as failed to re-build deduplication info.
@@ -21,13 +21,47 @@ Add a label named reason for topic_load_failed_total
 - label name: `reason`
 - label values:
   - `bundle_unloading`
-  - `failed_load_policies`
+  - `failed_load_namespace_policies`
+  - `failed_load_topic_policies`
   - `failed_load_ml`
+  - `failed_check_ownership`
   - `failed_access_metadata_store`
   - `failed_init`
-  - `timeout`
+  - `timeout` (generic timeout, or timeout while more than one loading stage is pending)
+  - `timeout_load_namespace_policies`
+  - `timeout_load_topic_policies`
+  - `timeout_load_ml`
+  - `timeout_init`
+  - `timeout_dedup`
   - `others`
 
+### Implementation Notes
+
+Each persistent topic-loading operation records at most one failure. The failure reason is attached at the
+operation that can identify it, rather than by wrapping exceptions solely for metrics. The final topic future
+increments the counter once, so callers that reuse an in-flight load future do not create additional samples.
+
+Timeouts describe the loading stage that is still pending when the timeout fires:
+
+| Pending stage | Timeout reason |
+|---|---|
+| namespace-level policy loading, including during `PersistentTopic` initialization | `timeout_load_namespace_policies` |
+| topic-level policy loading, including during `PersistentTopic` initialization | `timeout_load_topic_policies` |
+| opening the managed ledger | `timeout_load_ml` |
+| topic initialization, replication, or compacted-subscription setup | `timeout_init` |
+| deduplication initialization | `timeout_dedup` |
+| no single pending stage | `timeout` |
+
+`bundle_unloading` includes failures while a bundle is unloading and failures because this broker is not the
+topic owner. `TopicMigratedException` is an expected migration control flow and does not increment
+`topic_load_failed_total`.
+
+`failed_check_ownership` covers failures while resolving the namespace bundle or checking its ownership. A
+successful ownership check that determines this broker is not the owner is reported as `bundle_unloading`.
+
+`brk_topic_load_failed_count` remains available for compatibility and is the sum of all reason-labelled failure
+series.
+
 
 # Monitoring & Alternatives