[improve][build] Restore fastutil and minimize it in the client and server distributions

build-logic/conventions/src/main/kotlin/MinimizedDependenciesExtension.kt

@@ -0,0 +1,39 @@
+/*
+ * 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.
+ */
+
+import org.gradle.api.provider.ListProperty
+import org.gradle.api.provider.Property
+
+/**
+ * Configuration for the `pulsar.minimized-dependencies-conventions` plugin.
+ *
+ * A "minimized" packaging module declares its reachability roots as `api(project(...))`
+ * dependencies and then only needs to express which libraries to minimize and the
+ * expected upper bound on the retained class count.
+ */
+interface MinimizedDependenciesExtension {
+    /** Libraries to minimize, as `"group:name"` entries (e.g. `"it.unimi.dsi:fastutil"`). */
+    val minimizedDependencies: ListProperty<String>
+
+    /**
+     * Upper bound on the number of `.class` entries retained in the shaded jar. The build fails
+     * if the jar exceeds it — this catches a minimize() regression that would ship the full jar.
+     */
+    val maxRetainedClasses: Property<Int>
+}

build-logic/conventions/src/main/kotlin/pulsar.client-shade-conventions.gradle.kts

@@ -37,6 +37,7 @@ tasks.named<com.github.jengelman.gradle.plugins.shadow.tasks.ShadowJar>("shadowJ
         include(project(":pulsar-client-admin-original"))
         include(project(":pulsar-common"))
         include(project(":pulsar-client-messagecrypto-bc"))
+        include(project(":pulsar-client-fastutil-minimized"))
         include(dependency("com.fasterxml.jackson.*:.*"))
         include(dependency("com.google.*:.*"))
         include(dependency("com.google.auth:.*"))
@@ -143,6 +144,7 @@ tasks.named<com.github.jengelman.gradle.plugins.shadow.tasks.ShadowJar>("shadowJ
     relocateWithPrefix(shadePrefix, "io.opencensus")
     relocateWithPrefix(shadePrefix, "io.prometheus.client")
     relocateWithPrefix(shadePrefix, "io.swagger")
+    relocateWithPrefix(shadePrefix, "it.unimi.dsi.fastutil")
     relocateWithPrefix(shadePrefix, "javassist")
     relocateWithPrefix(shadePrefix, "jakarta.activation")
     relocateWithPrefix(shadePrefix, "jakarta.annotation")

build-logic/conventions/src/main/kotlin/pulsar.minimized-dependencies-conventions.gradle.kts

@@ -0,0 +1,93 @@
+/*
+ * 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.
+ */
+
+// Imported explicitly: the `java` plugin contributes a `java { }` extension accessor,
+// so an unqualified `java.util.zip.ZipFile` would resolve `java` to that extension.
+import com.github.jengelman.gradle.plugins.shadow.tasks.ShadowJar
+import java.util.zip.ZipFile
+
+// Convention for "<library> minimized" packaging modules. Produces a shadow jar that
+// contains only the classes of the minimized libraries that are actually reachable from
+// the module's reachability roots and their transitive closure.
+//
+// A consuming module:
+//   * applies this plugin,
+//   * declares its reachability roots as `api(project(...))` dependencies (e.g. the
+//     pulsar projects that use the libraries being minimized), and
+//   * configures `minimizedJar { minimizedDependencies.set(listOf("group:name")); maxRetainedClasses.set(N) }`.
+//
+// Why `api`: Shadow's minimize() seeds its reachability analysis (UnusedTracker) from
+// the project's own source classes plus its `api`-scoped jars. A packaging module has no
+// source, so the `api` roots are what drive the analysis; with `implementation` (or no
+// roots) minimize() has nothing to start from and keeps the whole jar.
+
+plugins {
+    id("pulsar.java-conventions")
+    id("pulsar.shadow-conventions")
+}
+
+val minimized = extensions.create<MinimizedDependenciesExtension>("minimizedJar")
+
+// The `api` roots are a build-only reachability seed; strip them (and everything else)
+// from the consumable variants so this module ships a self-contained jar with no
+// transitive dependencies on consumers' classpaths.
+listOf("apiElements", "runtimeElements").forEach { variant ->
+    configurations.named(variant) {
+        setExtendsFrom(emptySet())
+    }
+}
+
+tasks.named<ShadowJar>("shadowJar") {
+    val minimizedDeps = minimized.minimizedDependencies
+    inputs.property("minimizedDependencies", minimizedDeps)
+    // Bundle ONLY the minimized libraries; the `api` roots are read by minimize() as
+    // reachability roots but are not part of the output jar.
+    dependencies {
+        minimizedDeps.get().forEach { coords -> include(dependency("$coords:.*")) }
+    }
+    // Drop every bundled class not reachable from the reachability roots.
+    minimize()
+}
+
+// Verify the jar was actually pruned: the reachable set is small, so a count well above
+// it but far below the full library jar catches a minimize() regression (e.g. the no-op
+// that ships the whole jar). Stays configuration-cache compatible — the action captures
+// only Providers, never the Project.
+val verifyMinimizedJar = tasks.register("verifyMinimizedJar") {
+    val jarFile = tasks.named<ShadowJar>("shadowJar").flatMap { it.archiveFile }
+    val maxClasses = minimized.maxRetainedClasses
+    inputs.file(jarFile)
+    inputs.property("maxRetainedClasses", maxClasses)
+    doLast {
+        val limit = maxClasses.get()
+        val classCount = ZipFile(jarFile.get().asFile).use { zf ->
+            zf.entries().asSequence().count { it.name.endsWith(".class") }
+        }
+        if (classCount > limit) {
+            throw GradleException(
+                "Minimized jar retained $classCount classes (> $limit) — minimize() is not pruning."
+            )
+        }
+        logger.lifecycle("Minimized jar OK: $classCount classes retained (limit $limit).")
+    }
+}
+
+tasks.named("check") {
+    dependsOn(verifyMinimizedJar)
+}

distribution/server/build.gradle.kts

@@ -71,6 +71,9 @@ val distLib by configurations.creating {
     exclude(group = "com.google.android", module = "annotations")
     // Annotation libraries not needed at runtime
     exclude(group = "org.codehaus.mojo", module = "animal-sniffer-annotations")
+    // The full fastutil jar (~25MB) is replaced by :pulsar-broker-fastutil-minimized below,
+    // which ships only the fastutil classes actually used on the server (and client) side.
+    exclude(group = "it.unimi.dsi", module = "fastutil")
 }
 
 // Resolvable configurations for cross-project artifact dependencies.
@@ -91,6 +94,9 @@ dependencies {
     // Version constraints from the enforced platform (inherited via implementation,
     // which distLib extends) ensure consistent versions without manual resolutionStrategy.
     distLib(project(":pulsar-broker"))
+    // Minimized fastutil (replaces the full fastutil jar excluded from distLib above): only the
+    // fastutil classes reachable from the broker and the bundled pulsar-client-original.
+    distLib(project(":pulsar-broker-fastutil-minimized"))
     distLib(project(":pulsar-metadata"))
     distLib(project(":pulsar-docs-tools"))
     distLib(project(":pulsar-proxy"))

distribution/server/src/assemble/LICENSE.bin.txt

@@ -262,6 +262,7 @@ The Apache Software License, Version 2.0
      - com.fasterxml.jackson.module-jackson-module-parameter-names-2.21.3.jar
  * Caffeine -- com.github.ben-manes.caffeine-caffeine-3.2.4.jar
  * Conscrypt -- org.conscrypt-conscrypt-openjdk-uber-2.5.2.jar
+ * Fastutil -- it.unimi.dsi:fastutil (only the classes used by Pulsar, bundled within pulsar-broker-fastutil-minimized)
  * LMAX Disruptor -- com.lmax-disruptor-4.0.0.jar
  * Proto Google Common Protos -- com.google.api.grpc-proto-google-common-protos-2.63.2.jar
  * Bitbucket -- org.bitbucket.b_c-jose4j-0.9.6.jar

distribution/shell/build.gradle.kts

@@ -42,9 +42,15 @@ val distLib by configurations.creating {
     exclude(group = "net.java.dev.jna", module = "jna-platform")
     exclude(group = "io.netty", module = "netty-transport-native-kqueue")
     exclude(group = "io.prometheus", module = "simpleclient_caffeine")
+    // The full fastutil jar (~24MB) is replaced by :pulsar-client-fastutil-minimized below,
+    // which ships only the fastutil classes the client-side modules actually use.
+    exclude(group = "it.unimi.dsi", module = "fastutil")
 }
 dependencies {
     distLib(project(":pulsar-client-tools"))
+    // Minimized fastutil (replaces the full fastutil jar excluded from distLib above): only the
+    // fastutil classes reachable from the client-side modules (client, client-tools, admin).
+    distLib(project(":pulsar-client-fastutil-minimized"))
     distLib(libs.log4j.core)
     distLib(libs.log4j.web)
     distLib(libs.log4j.layout.template.json)

distribution/shell/src/assemble/LICENSE.bin.txt

@@ -427,6 +427,7 @@ The Apache Software License, Version 2.0
  * RE2j -- re2j-1.8.jar
  * Spotify completable-futures -- completable-futures-0.3.6.jar
  * RoaringBitmap -- RoaringBitmap-1.6.9.jar
+ * Fastutil -- it.unimi.dsi:fastutil (only the classes used by Pulsar, bundled within pulsar-client-fastutil-minimized)
  * JSpecify -- jspecify-1.0.0.jar
  * JetBrains Annotations -- annotations-26.1.0.jar
 

gradle/libs.versions.toml

@@ -80,6 +80,7 @@ opentelemetry-gcp-resources = "1.57.0-alpha"
 # Data structures / Utils
 guava = "33.6.0-jre"
 caffeine = "3.2.4"
+fastutil = "8.5.18"
 jctools = "4.0.6"
 roaringbitmap = "1.6.9"
 hppc = "0.9.1"
@@ -175,7 +176,7 @@ thrift = "0.23.0"
 datasketches-memory = "4.1.0"
 datasketches-java = "7.0.1"
 # Shading
-shadow = "9.4.1"
+shadow = "9.4.2"
 
 [libraries]
 # SLF4J
@@ -330,6 +331,7 @@ rocksdbjni = { module = "org.rocksdb:rocksdbjni", version.ref = "rocksdb" }
 error-prone-annotations = { module = "com.google.errorprone:error_prone_annotations", version.ref = "errorprone" }
 # Data structures
 caffeine = { module = "com.github.ben-manes.caffeine:caffeine", version.ref = "caffeine" }
+fastutil = { module = "it.unimi.dsi:fastutil", version.ref = "fastutil" }
 jctools-core = { module = "org.jctools:jctools-core", version.ref = "jctools" }
 jctools-core-jdk11 = { module = "org.jctools:jctools-core-jdk11", version.ref = "jctools" }
 roaringbitmap = { module = "org.roaringbitmap:RoaringBitmap", version.ref = "roaringbitmap" }

pulsar-broker-fastutil-minimized/build.gradle.kts

@@ -0,0 +1,42 @@
+/*
+ * 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.
+ */
+
+// Jar of just the fastutil classes reachable from the Pulsar server side. It replaces the
+// full fastutil jar in the server distribution so the docker image / tarball ships only the
+// classes actually used. The roots cover both the broker and (since it is bundled in the
+// server distribution) the unrelocated pulsar-client-original, so this is a superset of
+// :pulsar-client-fastutil-minimized. See pulsar.minimized-dependencies-conventions.
+
+plugins {
+    id("pulsar.minimized-dependencies-conventions")
+}
+
+dependencies {
+    // Reachability roots: every Pulsar project that uses fastutil and ends up in the server
+    // distribution. minimize() keeps the union of fastutil classes reachable from these.
+    api(project(":pulsar-broker"))
+    api(project(":pulsar-client-original"))
+}
+
+minimizedJar {
+    minimizedDependencies.set(listOf("it.unimi.dsi:fastutil"))
+    // The reachable set (broker + client usage) is ~818 classes; fail the build if it grows
+    // past this (bump it when new fastutil usage legitimately enlarges the set).
+    maxRetainedClasses.set(850)
+}

pulsar-broker/build.gradle.kts

@@ -49,6 +49,7 @@ dependencies {
     api(libs.commons.lang3)
     api(libs.netty.transport)
     implementation(libs.protobuf.java)
+    implementation(libs.fastutil)
     implementation(libs.curator.recipes)
     implementation(libs.bookkeeper.stream.storage.server) {
         exclude(group = "org.apache.bookkeeper")

pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTracker.java

@@ -21,17 +21,21 @@
 import com.google.common.annotations.VisibleForTesting;
 import io.github.merlimat.slog.Logger;
 import io.netty.util.Timer;
+import it.unimi.dsi.fastutil.longs.Long2ObjectAVLTreeMap;
+import it.unimi.dsi.fastutil.longs.Long2ObjectMap;
+import it.unimi.dsi.fastutil.longs.Long2ObjectRBTreeMap;
+import it.unimi.dsi.fastutil.longs.Long2ObjectSortedMap;
+import it.unimi.dsi.fastutil.longs.LongOpenHashSet;
+import it.unimi.dsi.fastutil.longs.LongSet;
 import java.time.Clock;
 import java.util.NavigableSet;
-import java.util.TreeMap;
 import java.util.TreeSet;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicLong;
 import lombok.Getter;
 import org.apache.bookkeeper.mledger.Position;
 import org.apache.bookkeeper.mledger.PositionFactory;
 import org.apache.pulsar.broker.service.persistent.AbstractPersistentDispatcherMultipleConsumers;
-import org.apache.pulsar.common.util.collections.LongOpenHashSet;
 import org.roaringbitmap.longlong.Roaring64Bitmap;
 
 public class InMemoryDelayedDeliveryTracker extends AbstractDelayedDeliveryTracker {
@@ -40,9 +44,9 @@ public class InMemoryDelayedDeliveryTracker extends AbstractDelayedDeliveryTrack
     protected final Logger log;
 
     // timestamp -> ledgerId -> entryId
-    // TreeMap -> TreeMap -> RoaringBitmap
-    protected final TreeMap<Long, TreeMap<Long, Roaring64Bitmap>>
-            delayedMessageMap = new TreeMap<>();
+    // AVL tree -> OpenHashMap -> RoaringBitmap
+    protected final Long2ObjectSortedMap<Long2ObjectSortedMap<Roaring64Bitmap>>
+            delayedMessageMap = new Long2ObjectAVLTreeMap<>();
 
     // If we detect that all messages have fixed delay time, such that the delivery is
     // always going to be in FIFO order, then we can avoid pulling all the messages in
@@ -137,7 +141,7 @@ public boolean addMessage(long ledgerId, long entryId, long deliverAt) {
                 .log("Add message");
         long timestamp = roundTimestamp(deliverAt);
 
-        Roaring64Bitmap bitmap = delayedMessageMap.computeIfAbsent(timestamp, k -> new TreeMap<>())
+        Roaring64Bitmap bitmap = delayedMessageMap.computeIfAbsent(timestamp, k -> new Long2ObjectRBTreeMap<>())
             .computeIfAbsent(ledgerId, k -> new Roaring64Bitmap());
         // Roaring64Bitmap does not store duplicates, so track if it a new element
         // so we can keep delayedMessagesCount in sync
@@ -194,7 +198,7 @@ private void checkAndUpdateHighest(long deliverAt) {
     @Override
     public boolean hasMessageAvailable() {
         boolean hasMessageAvailable = !delayedMessageMap.isEmpty()
-                && delayedMessageMap.firstKey() <= getCutoffTime();
+                && delayedMessageMap.firstLongKey() <= getCutoffTime();
         if (!hasMessageAvailable) {
             updateTimer();
         }
@@ -211,15 +215,15 @@ public NavigableSet<Position> getScheduledMessages(int maxMessages) {
         long cutoffTime = getCutoffTime();
 
         while (n > 0 && !delayedMessageMap.isEmpty()) {
-            long timestamp = delayedMessageMap.firstKey();
+            long timestamp = delayedMessageMap.firstLongKey();
             if (timestamp > cutoffTime) {
                 break;
             }
 
-            LongOpenHashSet ledgerIdToDelete = new LongOpenHashSet();
-            TreeMap<Long, Roaring64Bitmap> ledgerMap = delayedMessageMap.get(timestamp);
-            for (var ledgerEntry : ledgerMap.entrySet()) {
-                long ledgerId = ledgerEntry.getKey();
+            LongSet ledgerIdToDelete = new LongOpenHashSet();
+            Long2ObjectSortedMap<Roaring64Bitmap> ledgerMap = delayedMessageMap.get(timestamp);
+            for (Long2ObjectMap.Entry<Roaring64Bitmap> ledgerEntry : ledgerMap.long2ObjectEntrySet()) {
+                long ledgerId = ledgerEntry.getLongKey();
                 Roaring64Bitmap entryIds = ledgerEntry.getValue();
                 long cardinality = entryIds.getLongCardinality();
                 if (cardinality <= n) {
@@ -309,6 +313,6 @@ && getNumberOfDelayedMessages() >= fixedDelayDetectionLookahead
     }
 
     protected long nextDeliveryTime() {
-        return delayedMessageMap.firstKey();
+        return delayedMessageMap.firstLongKey();
     }
 }

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java

@@ -28,6 +28,8 @@
 import io.netty.util.concurrent.Future;
 import io.netty.util.concurrent.Promise;
 import io.opentelemetry.api.common.Attributes;
+import it.unimi.dsi.fastutil.ints.IntIntPair;
+import it.unimi.dsi.fastutil.objects.ObjectIntPair;
 import java.time.Instant;
 import java.util.ArrayList;
 import java.util.BitSet;
@@ -72,8 +74,6 @@
 import org.apache.pulsar.common.util.DateFormatter;
 import org.apache.pulsar.common.util.FutureUtil;
 import org.apache.pulsar.common.util.collections.BitSetRecyclable;
-import org.apache.pulsar.common.util.collections.IntIntPair;
-import org.apache.pulsar.common.util.collections.ObjectIntPair;
 import org.apache.pulsar.opentelemetry.OpenTelemetryAttributes;
 import org.apache.pulsar.transaction.common.exception.TransactionConflictException;
 

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/DrainingHashesTracker.java

@@ -19,6 +19,7 @@
 package org.apache.pulsar.broker.service;
 
 import static org.apache.pulsar.broker.service.StickyKeyConsumerSelector.STICKY_KEY_HASH_NOT_SET;
+import it.unimi.dsi.fastutil.ints.Int2ObjectOpenHashMap;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
@@ -32,7 +33,6 @@
 import org.apache.pulsar.common.policies.data.DrainingHash;
 import org.apache.pulsar.common.policies.data.stats.ConsumerStatsImpl;
 import org.apache.pulsar.common.policies.data.stats.DrainingHashImpl;
-import org.apache.pulsar.common.util.collections.Int2ObjectOpenHashMap;
 import org.roaringbitmap.RoaringBitmap;
 
 /**