Sync master-aligned delayed delivery & cursor persistence work into branch-as-4.0 (CI verification)

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

@@ -67,7 +67,7 @@ jobs:
     runs-on: ubuntu-24.04
     strategy:
       matrix:
-        go-version: ['1.24']
+        go-version: ['1.25']
 
     steps:
       - name: Check out code into the Go module directory
@@ -85,7 +85,7 @@ jobs:
       - name: InstallTool
         run: |
           cd pulsar-function-go
-          wget -O - -q https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh| sh -s v2.0.2
+          wget -O - -q https://raw.githubusercontent.com/golangci/golangci-lint/main/install.sh| sh -s v2.12.2
           ./bin/golangci-lint --version
 
       - name: Build

.github/workflows/pulsar-ci.yaml

@@ -537,7 +537,7 @@ jobs:
           $GITHUB_WORKSPACE/build/pulsar_ci_tool.sh restore_tar_from_github_actions_artifacts pulsar-maven-repository-binaries
 
       - name: Set up QEMU
-        uses: docker/setup-qemu-action@29109295f81e9208d7d86ff1c6c12d2833863392
+        uses: docker/setup-qemu-action@06116385d9baf250c9f4dcb4858b16962ea869c3 # v4.1.0
         with:
           platforms: arm64
 
@@ -944,25 +944,26 @@ jobs:
       - name: Check binary licenses
         run: src/check-binary-license.sh ./distribution/server/target/apache-pulsar-*-bin.tar.gz && src/check-binary-license.sh ./distribution/shell/target/apache-pulsar-shell-*-bin.tar.gz
 
-#      - name: Run Trivy container scan
-#        id: trivy_scan
-#        uses: aquasecurity/trivy-action@v0.35.0
-#        if: ${{ github.repository == 'apache/pulsar' && github.event_name != 'pull_request' }}
-#        continue-on-error: true
-#        with:
-#          image-ref: "apachepulsar/pulsar:latest"
-#          scanners: vuln
-#          severity: CRITICAL,HIGH,MEDIUM,LOW
-#          limit-severities-for-sarif: true
-#          format: 'sarif'
-#          output: 'trivy-results.sarif'
-#
-#      - name: Upload Trivy scan results to GitHub Security tab
-#        uses: github/codeql-action/upload-sarif@v3
-#        if: ${{ steps.trivy_scan.outcome == 'success' && github.repository == 'apache/pulsar' && github.event_name != 'pull_request' }}
-#        continue-on-error: true
-#        with:
-#          sarif_file: 'trivy-results.sarif'
+      - name: Run Trivy container scan
+        id: trivy_scan
+        uses: lhotari/sandboxed-trivy-action@f01374b6cc3bf7264ab238293e94f6db7ada6dd0
+        if: ${{ github.repository == 'apache/pulsar' && github.event_name != 'pull_request' }}
+        continue-on-error: true
+        with:
+          scan-type: 'image'
+          scan-ref: "apachepulsar/pulsar:latest"
+          scanners: vuln
+          severity: CRITICAL,HIGH,MEDIUM,LOW
+          limit-severities-for-sarif: true
+          format: 'sarif'
+          output: 'trivy-results.sarif'
+
+      - name: Upload Trivy scan results to GitHub Security tab
+        uses: github/codeql-action/upload-sarif@v4
+        if: ${{ steps.trivy_scan.outcome == 'success' && github.repository == 'apache/pulsar' && github.event_name != 'pull_request' }}
+        continue-on-error: true
+        with:
+          sarif_file: 'trivy-results.sarif'
 
       - name: Clean up disk space
         run: |

bin/pulsar-admin-common.sh

@@ -127,6 +127,14 @@ if [[ $JAVA_MAJOR_VERSION -ge 11 ]]; then
   OPTS="$OPTS --add-opens java.base/java.nio=ALL-UNNAMED --add-opens java.base/jdk.internal.misc=ALL-UNNAMED"
 fi
 
+if [[ $JAVA_MAJOR_VERSION -ge 24 ]]; then
+  # Netty loads native libraries (epoll, io_uring, tcnative) via java.lang.System::loadLibrary,
+  # which is a restricted method from Java 24 onwards. Without this the JVM prints a warning to
+  # stderr on every invocation, and restricted methods will be blocked outright in a future
+  # release. bin/pulsar already sets this for the server side.
+  OPTS="$OPTS --enable-native-access=ALL-UNNAMED"
+fi
+
 OPTS="-cp $PULSAR_CLASSPATH $OPTS"
 
 OPTS="$OPTS $PULSAR_EXTRA_OPTS"

bin/pulsar-perf

@@ -124,6 +124,14 @@ if [[ $JAVA_MAJOR_VERSION -ge 11 ]]; then
   OPTS="$OPTS --add-opens java.base/java.nio=ALL-UNNAMED --add-opens java.base/jdk.internal.misc=ALL-UNNAMED"
 fi
 
+if [[ $JAVA_MAJOR_VERSION -ge 24 ]]; then
+  # Netty loads native libraries (epoll, io_uring, tcnative) via java.lang.System::loadLibrary,
+  # which is a restricted method from Java 24 onwards. Without this the JVM prints a warning to
+  # stderr on every invocation, and restricted methods will be blocked outright in a future
+  # release. bin/pulsar already sets this for the server side.
+  OPTS="$OPTS --enable-native-access=ALL-UNNAMED"
+fi
+
 OPTS="-cp $PULSAR_CLASSPATH $OPTS"
 OPTS="$OPTS $PULSAR_EXTRA_OPTS"
 

bouncy-castle/bc/pom.xml

@@ -25,7 +25,7 @@
   <parent>
     <groupId>com.ascentstream.pulsar</groupId>
     <artifactId>bouncy-castle-parent</artifactId>
-    <version>4.0.11.1-226281f-20260709031812-SNAPSHOT</version>
+    <version>4.0.14-SNAPSHOT</version>
   </parent>
 
   <artifactId>bouncy-castle-bc</artifactId>

bouncy-castle/bcfips-include-test/pom.xml

@@ -24,7 +24,7 @@
   <parent>
     <groupId>com.ascentstream.pulsar</groupId>
     <artifactId>bouncy-castle-parent</artifactId>
-    <version>4.0.11.1-226281f-20260709031812-SNAPSHOT</version>
+    <version>4.0.14-SNAPSHOT</version>
   </parent>
 
   <artifactId>bcfips-include-test</artifactId>

bouncy-castle/bcfips/pom.xml

@@ -25,7 +25,7 @@
   <parent>
     <groupId>com.ascentstream.pulsar</groupId>
     <artifactId>bouncy-castle-parent</artifactId>
-    <version>4.0.11.1-226281f-20260709031812-SNAPSHOT</version>
+    <version>4.0.14-SNAPSHOT</version>
   </parent>
 
   <artifactId>bouncy-castle-bcfips</artifactId>

bouncy-castle/pom.xml

@@ -25,7 +25,7 @@
   <parent>
     <groupId>com.ascentstream.pulsar</groupId>
     <artifactId>pulsar</artifactId>
-    <version>4.0.11.1-226281f-20260709031812-SNAPSHOT</version>
+    <version>4.0.14-SNAPSHOT</version>
   </parent>
 
   <build>

buildtools/pom.xml

@@ -31,7 +31,7 @@
 
   <groupId>com.ascentstream.pulsar</groupId>
   <artifactId>buildtools</artifactId>
-  <version>4.0.11.1-226281f-20260709031812-SNAPSHOT</version>
+  <version>4.0.14-SNAPSHOT</version>
   <packaging>jar</packaging>
   <name>Pulsar Build Tools</name>
 

buildtools/src/main/resources/log4j2.xml

@@ -22,7 +22,7 @@
 <Configuration status="INFO">
     <Appenders>
         <Console name="Console" target="SYSTEM_OUT">
-            <PatternLayout pattern="%d{ISO8601_OFFSET_DATE_TIME_HHMM} %-5level [%t{12}] %c{1.} - %msg %X%n" />
+            <PatternLayout pattern="%d{ISO8601_OFFSET_DATE_TIME_HHMM} %-5level [%t{12}] %c{1.} - %msg%equals{ %X}{ {}}{}%n" />
         </Console>
     </Appenders>
     <Loggers>

conf/broker.conf

@@ -769,6 +769,13 @@ systemTopicSchemaCompatibilityStrategy=ALWAYS_COMPATIBLE
 # Please enable the system topic first.
 topicLevelPoliciesEnabled=true
 
+# Amount of seconds to timeout initializing the topic policies cache of a namespace (reading the namespace's
+# __change_events system topic to the end). Topic loading waits for this initialization, so if the system-topic
+# reader gets stuck, this bounds the wait: the broker fails the initialization, closes the stuck reader and clears
+# the cached state so loading the namespace's topics can be retried with a fresh reader instead of hanging until the
+# broker is restarted. Set to 0 or a negative value to disable the timeout (not recommended).
+topicPoliciesCacheInitTimeoutSeconds=60
+
 # If a topic remains fenced for this number of seconds, it will be closed forcefully.
 # If it is set to 0 or a negative number, the fenced topic will not be closed.
 topicFencingTimeoutSeconds=0
@@ -1372,15 +1379,9 @@ managedLedgerMaxUnackedRangesToPersist=10000
 managedLedgerMaxBatchDeletedIndexToPersist=10000
 
 # When storing acknowledgement state, choose a more compact serialization format that stores
-# individual acknowledgements as a bitmap which is serialized to an array of long values. NOTE: This setting requires
-# managedLedgerUnackedRangesOpenCacheSetEnabled=true to be effective.
+# individual acknowledgements as a bitmap which is serialized to an array of long values.
 managedLedgerPersistIndividualAckAsLongArray=false
 
-# When set to true, a BitSet will be used to track acknowledged messages that come after the "mark delete position"
-# for each subscription. RoaringBitmap is used as a memory efficient BitSet implementation for the acknowledged
-# messages tracking. Unacknowledged ranges are the message ranges excluding the acknowledged messages.
-managedLedgerUnackedRangesOpenCacheSetEnabled=true
-
 # Max number of "acknowledgment holes" that can be stored in MetadataStore. If number of unack message range is higher
 # than this limit then broker will persist unacked ranges into bookkeeper to avoid additional data overhead into
 # MetadataStore.
@@ -1805,6 +1806,12 @@ metricsServletTimeoutMs=30000
 # Enable or disable broker bundles metrics. The default value is false.
 exposeBundlesMetricsInPrometheus=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
+
 ### --- Functions --- ###
 
 # Enable Functions Worker Service in Broker
@@ -2078,6 +2085,12 @@ dispatcherPauseOnAckStatePersistentEnabled=false
 # multiple entries.
 persistentUnackedRangesWithMultipleEntriesEnabled=false
 
+# Enables per-msgLedger cursor checkpoint persistence. When true, the cursor writes a
+# CursorCheckpoint per flush (mark-delete ledger's ack state inline + refs to other ledgers'
+# previously-persisted ack states) instead of the legacy single PositionInfo entry. Eliminates
+# write amplification and avoids ack truncation.
+persistentUnackedRangesWithPerLedgerEntryEnabled=false
+
 # Deprecated - Use managedLedgerCacheEvictionIntervalMs instead
 managedLedgerCacheEvictionFrequency=0
 

conf/functions_log4j2.xml

@@ -40,7 +40,7 @@
             <name>Console</name>
             <target>SYSTEM_OUT</target>
             <PatternLayout>
-                <Pattern>%d{ISO8601_OFFSET_DATE_TIME_HHMM} [%t] %-5level %logger{36} - %msg%n</Pattern>
+                <Pattern>%d{ISO8601_OFFSET_DATE_TIME_HHMM} [%t] %-5level %logger{36} - %msg%equals{ %X}{ {}}{}%n</Pattern>
             </PatternLayout>
         </Console>
         <RollingFile>
@@ -49,7 +49,7 @@
             <filePattern>${sys:pulsar.function.log.dir}/${sys:pulsar.function.log.file}-%d{MM-dd-yyyy}-%i.log.gz</filePattern>
             <immediateFlush>true</immediateFlush>
             <PatternLayout>
-                <Pattern>%d{ISO8601_OFFSET_DATE_TIME_HHMM} [%t] %-5level %logger{36} - %msg%n</Pattern>
+                <Pattern>%d{ISO8601_OFFSET_DATE_TIME_HHMM} [%t] %-5level %logger{36} - %msg%equals{ %X}{ {}}{}%n</Pattern>
             </PatternLayout>
             <Policies>
                 <TimeBasedTriggeringPolicy>
@@ -82,7 +82,7 @@
             <filePattern>${sys:pulsar.function.log.dir}/${sys:pulsar.function.log.file}.bk-%d{MM-dd-yyyy}-%i.log.gz</filePattern>
             <immediateFlush>true</immediateFlush>
             <PatternLayout>
-                <Pattern>%d{ISO8601_OFFSET_DATE_TIME_HHMM} [%t] %-5level %logger{36} - %msg%n</Pattern>
+                <Pattern>%d{ISO8601_OFFSET_DATE_TIME_HHMM} [%t] %-5level %logger{36} - %msg%equals{ %X}{ {}}{}%n</Pattern>
             </PatternLayout>
             <Policies>
                 <TimeBasedTriggeringPolicy>

conf/functions_worker.yml

@@ -198,6 +198,9 @@ functionRuntimeFactoryConfigs:
 #    # The Kubernetes pod name to run the function instances. It is set to
 #    # `pf-<tenant>-<namespace>-<function_name>-<random_uuid(8)>` if this setting is left to be empty
 #    jobName:
+#    # Optional domain suffix to use when the Function Worker constructs the gRPC address to connect to function instances.
+#    # If left blank, it defaults to `.svc.cluster.local`. Set this if your Function Worker is outside the cluster and connects via an external Gateway/Ingress.
+#    kubernetesServiceDomainSuffix:
 #    # the docker image to run function instance. by default it is `apachepulsar/pulsar`
 #    pulsarDockerImageName:
 #    # the docker image to run function instance according to different configurations provided by users.

conf/log4j2.yaml

@@ -58,7 +58,7 @@ Configuration:
       - name: Console
         target: SYSTEM_OUT
         PatternLayout:
-          Pattern: "%d{ISO8601_OFFSET_DATE_TIME_HHMM} [%t] %-5level %logger{36} - %msg%n"
+          Pattern: "%d{ISO8601_OFFSET_DATE_TIME_HHMM} [%t] %-5level %logger{36} - %msg%equals{ %X}{ {}}{}%n"
       - name: ConsoleJson
         target: SYSTEM_OUT
         JsonTemplateLayout:
@@ -71,7 +71,7 @@ Configuration:
       filePattern: "${sys:pulsar.log.dir}/${sys:pulsar.log.file}-%d{MM-dd-yyyy}-%i.log.gz"
       immediateFlush: ${sys:pulsar.log.immediateFlush}
       PatternLayout:
-        Pattern: "%d{ISO8601_OFFSET_DATE_TIME_HHMM} [%t] %-5level %logger{36} - %msg%n"
+        Pattern: "%d{ISO8601_OFFSET_DATE_TIME_HHMM} [%t] %-5level %logger{36} - %msg%equals{ %X}{ {}}{}%n"
       Policies:
         TimeBasedTriggeringPolicy:
           interval: 1
@@ -109,7 +109,7 @@ Configuration:
                       fileName : "${sys:pulsar.log.dir}/functions/${ctx:function}/${ctx:functionname}-${ctx:instance}.log"
                       filePattern : "${sys:pulsar.log.dir}/functions/${sys:pulsar.log.file}-${ctx:instance}-%d{MM-dd-yyyy}-%i.log.gz"
                       PatternLayout:
-                        Pattern: "%d{ABSOLUTE} %level{length=5} [%thread] [instance: %X{instance}] %logger{1} - %msg%n"
+                        Pattern: "%d{ABSOLUTE} %level{length=5} [%thread] [instance:%X{instance}] %logger{1} - %msg%equals{ %X}{ {}}{}%n"
                       Policies:
                         TimeBasedTriggeringPolicy:
                           interval: 1

conf/standalone.conf

@@ -579,6 +579,13 @@ systemTopicSchemaCompatibilityStrategy=ALWAYS_COMPATIBLE
 # Please enable the system topic first.
 topicLevelPoliciesEnabled=true
 
+# Amount of seconds to timeout initializing the topic policies cache of a namespace (reading the namespace's
+# __change_events system topic to the end). Topic loading waits for this initialization, so if the system-topic
+# reader gets stuck, this bounds the wait: the broker fails the initialization, closes the stuck reader and clears
+# the cached state so loading the namespace's topics can be retried with a fresh reader instead of hanging until the
+# broker is restarted. Set to 0 or a negative value to disable the timeout (not recommended).
+topicPoliciesCacheInitTimeoutSeconds=60
+
 # If a topic remains fenced for this number of seconds, it will be closed forcefully.
 # If it is set to 0 or a negative number, the fenced topic will not be closed.
 topicFencingTimeoutSeconds=0
@@ -868,15 +875,9 @@ managedLedgerMaxUnackedRangesToPersist=10000
 managedLedgerMaxBatchDeletedIndexToPersist=10000
 
 # When storing acknowledgement state, choose a more compact serialization format that stores
-# individual acknowledgements as a bitmap which is serialized to an array of long values. NOTE: This setting requires
-# managedLedgerUnackedRangesOpenCacheSetEnabled=true to be effective.
+# individual acknowledgements as a bitmap which is serialized to an array of long values.
 managedLedgerPersistIndividualAckAsLongArray=false
 
-# When set to true, a BitSet will be used to track acknowledged messages that come after the "mark delete position"
-# for each subscription. RoaringBitmap is used as a memory efficient BitSet implementation for the acknowledged
-# messages tracking. Unacknowledged ranges are the message ranges excluding the acknowledged messages.
-managedLedgerUnackedRangesOpenCacheSetEnabled=true
-
 # Max number of "acknowledgment holes" that can be stored in MetadataStore. If number of unack message range is higher
 # than this limit then broker will persist unacked ranges into bookkeeper to avoid additional data overhead into
 # MetadataStore.
@@ -1116,6 +1117,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).

deployment/terraform-ansible/templates/broker.conf

@@ -1077,9 +1077,6 @@ managedLedgerOffloadMaxThreads=2
 # Maximum prefetch rounds for ledger reading for offloading
 managedLedgerOffloadPrefetchRounds=1
 
-# Use Open Range-Set to cache unacked messages
-managedLedgerUnackedRangesOpenCacheSetEnabled=true
-
 # For Amazon S3 ledger offload, AWS region
 s3ManagedLedgerOffloadRegion=
 

distribution/io/pom.xml

@@ -25,7 +25,7 @@
   <parent>
     <groupId>com.ascentstream.pulsar</groupId>
     <artifactId>distribution</artifactId>
-    <version>4.0.11.1-226281f-20260709031812-SNAPSHOT</version>
+    <version>4.0.14-SNAPSHOT</version>
   </parent>
 
   <artifactId>pulsar-io-distribution</artifactId>

distribution/offloaders/pom.xml

@@ -25,7 +25,7 @@
   <parent>
     <groupId>com.ascentstream.pulsar</groupId>
     <artifactId>distribution</artifactId>
-    <version>4.0.11.1-226281f-20260709031812-SNAPSHOT</version>
+    <version>4.0.14-SNAPSHOT</version>
   </parent>
 
   <artifactId>pulsar-offloader-distribution</artifactId>

distribution/pom.xml

@@ -25,7 +25,7 @@
   <parent>
     <groupId>com.ascentstream.pulsar</groupId>
     <artifactId>pulsar</artifactId>
-    <version>4.0.11.1-226281f-20260709031812-SNAPSHOT</version>
+    <version>4.0.14-SNAPSHOT</version>
   </parent>
 
   <artifactId>distribution</artifactId>

distribution/server/pom.xml

@@ -25,7 +25,7 @@
   <parent>
     <groupId>com.ascentstream.pulsar</groupId>
     <artifactId>distribution</artifactId>
-    <version>4.0.11.1-226281f-20260709031812-SNAPSHOT</version>
+    <version>4.0.14-SNAPSHOT</version>
   </parent>
 
   <artifactId>pulsar-server-distribution</artifactId>
@@ -93,12 +93,6 @@
       <version>${project.version}</version>
     </dependency>
 
-    <dependency>
-      <groupId>jline</groupId>
-      <artifactId>jline</artifactId>
-      <version>${jline.version}</version>
-    </dependency>
-
     <dependency>
       <groupId>${project.groupId}</groupId>
       <artifactId>pulsar-package-bookkeeper-storage</artifactId>