[improve][client] PIP-393: Improve performance of Negative Acknowledgement

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

@@ -418,6 +418,8 @@ The Apache Software License, Version 2.0
     - avro-protobuf-1.11.4.jar
  * RE2j -- re2j-1.7.jar
  * Spotify completable-futures -- completable-futures-0.3.6.jar
+ * RoaringBitmap -- RoaringBitmap-1.2.0.jar
+ * Fastutil -- fastutil-8.5.14.jar
 
 BSD 3-clause "New" or "Revised" License
  * JSR305 -- jsr305-3.0.2.jar -- ../licenses/LICENSE-JSR305.txt

pom.xml

@@ -2586,6 +2586,7 @@ flexible messaging model and an intuitive client API.</description>
         <module>pulsar-metadata</module>
         <module>jetcd-core-shaded</module>
         <module>jclouds-shaded</module>
+        <module>pulsar-client-dependencies-minimized</module>
 
         <!-- package management releated modules (begin) -->
         <module>pulsar-package-management</module>
@@ -2651,6 +2652,7 @@ flexible messaging model and an intuitive client API.</description>
         <module>distribution</module>
         <module>pulsar-metadata</module>
         <module>jetcd-core-shaded</module>
+        <module>pulsar-client-dependencies-minimized</module>
         <!-- package management releated modules (begin) -->
         <module>pulsar-package-management</module>
         <!-- package management releated modules (end) -->

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/NegativeAcksTest.java

@@ -19,6 +19,7 @@
 package org.apache.pulsar.client.impl;
 
 import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
 import static org.testng.Assert.assertNull;
 import static org.testng.Assert.assertTrue;
 import java.util.HashSet;
@@ -311,19 +312,64 @@ public void testNegativeAcksDeleteFromUnackedTracker() throws Exception {
         // negative topic message id
         consumer.negativeAcknowledge(topicMessageId);
         NegativeAcksTracker negativeAcksTracker = consumer.getNegativeAcksTracker();
-        assertEquals(negativeAcksTracker.getNackedMessagesCount().orElse((long) -1).longValue(), 1L);
+        assertEquals(negativeAcksTracker.getNackedMessagesCount(), 1L);
         assertEquals(unAckedMessageTracker.size(), 0);
         negativeAcksTracker.close();
         // negative batch message id
         unAckedMessageTracker.add(messageId);
         consumer.negativeAcknowledge(batchMessageId);
         consumer.negativeAcknowledge(batchMessageId2);
         consumer.negativeAcknowledge(batchMessageId3);
-        assertEquals(negativeAcksTracker.getNackedMessagesCount().orElse((long) -1).longValue(), 1L);
+        assertEquals(negativeAcksTracker.getNackedMessagesCount(), 1L);
         assertEquals(unAckedMessageTracker.size(), 0);
         negativeAcksTracker.close();
     }
 
+    /**
+     * If we nack multiple messages in the same batch with different redelivery delays, the messages should be redelivered
+     * with the correct delay. However, all messages are redelivered at the same time.
+     * @throws Exception
+     */
+    @Test
+    public void testNegativeAcksWithBatch() throws Exception {
+        cleanup();
+        conf.setAcknowledgmentAtBatchIndexLevelEnabled(true);
+        setup();
+        String topic = BrokerTestUtil.newUniqueName("testNegativeAcksWithBatch");
+
+        @Cleanup
+        Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
+                .topic(topic)
+                .subscriptionName("sub1")
+                .acknowledgmentGroupTime(0, TimeUnit.SECONDS)
+                .subscriptionType(SubscriptionType.Shared)
+                .enableBatchIndexAcknowledgment(true)
+                .negativeAckRedeliveryDelay(3, TimeUnit.SECONDS)
+                .subscribe();
+
+        @Cleanup
+        Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+                .topic(topic)
+                .enableBatching(true)
+                .batchingMaxPublishDelay(1, TimeUnit.HOURS)
+                .batchingMaxMessages(2)
+                .create();
+        // send two messages in the same batch
+        producer.sendAsync("test-0");
+        producer.sendAsync("test-1");
+        producer.flush();
+
+        // negative ack the first message
+        consumer.negativeAcknowledge(consumer.receive());
+        // wait for 2s, negative ack the second message
+        Thread.sleep(2000);
+        consumer.negativeAcknowledge(consumer.receive());
+
+        // now 2s has passed, the first message should be redelivered 1s later.
+        Message<String> msg1 = consumer.receive(2, TimeUnit.SECONDS);
+        assertNotNull(msg1);
+    }
+
     @Test
     public void testNegativeAcksWithBatchAckEnabled() throws Exception {
         cleanup();

pulsar-client-admin-shaded/pom.xml

@@ -34,6 +34,17 @@
       <groupId>${project.groupId}</groupId>
       <artifactId>pulsar-client-admin-original</artifactId>
       <version>${project.version}</version>
+      <exclusions>
+        <exclusion>
+          <groupId>it.unimi.dsi</groupId>
+          <artifactId>fastutil</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>pulsar-client-dependencies-minimized</artifactId>
+      <version>${project.version}</version>
     </dependency>
     <dependency>
       <groupId>${project.groupId}</groupId>
@@ -150,6 +161,8 @@
                   <include>org.objenesis:*</include>
                   <include>org.reactivestreams:reactive-streams</include>
                   <include>org.yaml:snakeyaml</include>
+                  <include>org.apache.pulsar:pulsar-client-dependencies-minimized</include>
+                  <include>org.roaringbitmap:RoaringBitmap</include>
                 </includes>
                 <excludes>
                   <exclude>com.fasterxml.jackson.core:jackson-annotations</exclude>
@@ -269,6 +282,10 @@
                   <pattern>io.swagger</pattern>
                   <shadedPattern>org.apache.pulsar.shade.io.swagger</shadedPattern>
                 </relocation>
+                <relocation>
+                  <pattern>it.unimi.dsi.fastutil</pattern>
+                  <shadedPattern>org.apache.pulsar.shade.it.unimi.dsi.fastutil</shadedPattern>
+                </relocation>
                 <relocation>
                   <pattern>javassist</pattern>
                   <shadedPattern>org.apache.pulsar.shade.javassist</shadedPattern>
@@ -313,6 +330,11 @@
                   <shadedPattern>META-INF/versions/$1/org/apache/pulsar/shade/org/glassfish/</shadedPattern>
                   <rawString>true</rawString>
                 </relocation>
+                <relocation>
+                  <pattern>META-INF/versions/(\d+)/org/roaringbitmap/</pattern>
+                  <shadedPattern>META-INF/versions/$1/org/apache/pulsar/shade/org/roaringbitmap/</shadedPattern>
+                  <rawString>true</rawString>
+                </relocation>
                 <relocation>
                   <pattern>META-INF/versions/(\d+)/org/yaml/</pattern>
                   <shadedPattern>META-INF/versions/$1/org/apache/pulsar/shade/org/yaml/</shadedPattern>
@@ -374,6 +396,10 @@
                   <pattern>org.reactivestreams</pattern>
                   <shadedPattern>org.apache.pulsar.shade.org.reactivestreams</shadedPattern>
                 </relocation>
+                <relocation>
+                  <pattern>org.roaringbitmap</pattern>
+                  <shadedPattern>org.apache.pulsar.shade.org.roaringbitmap</shadedPattern>
+                </relocation>
                 <relocation>
                   <pattern>org.yaml</pattern>
                   <shadedPattern>org.apache.pulsar.shade.org.yaml</shadedPattern>

pulsar-client-all/pom.xml

@@ -39,6 +39,17 @@
       <groupId>${project.groupId}</groupId>
       <artifactId>pulsar-client-original</artifactId>
       <version>${project.version}</version>
+      <exclusions>
+        <exclusion>
+          <groupId>it.unimi.dsi</groupId>
+          <artifactId>fastutil</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>pulsar-client-dependencies-minimized</artifactId>
+      <version>${project.version}</version>
     </dependency>
     <dependency>
       <groupId>${project.groupId}</groupId>
@@ -200,6 +211,8 @@
                   <include>org.reactivestreams:reactive-streams</include>
                   <include>org.tukaani:xz</include>
                   <include>org.yaml:snakeyaml</include>
+                  <include>org.apache.pulsar:pulsar-client-dependencies-minimized</include>
+                  <include>org.roaringbitmap:RoaringBitmap</include>
                 </includes>
                 <excludes>
                   <exclude>com.fasterxml.jackson.core:jackson-annotations</exclude>
@@ -317,6 +330,10 @@
                   <pattern>io.swagger</pattern>
                   <shadedPattern>org.apache.pulsar.shade.io.swagger</shadedPattern>
                 </relocation>
+                <relocation>
+                  <pattern>it.unimi.dsi.fastutil</pattern>
+                  <shadedPattern>org.apache.pulsar.shade.it.unimi.dsi.fastutil</shadedPattern>
+                </relocation>
                 <relocation>
                   <pattern>javassist</pattern>
                   <shadedPattern>org.apache.pulsar.shade.javassist</shadedPattern>
@@ -361,6 +378,11 @@
                   <shadedPattern>META-INF/versions/$1/org/apache/pulsar/shade/org/glassfish/</shadedPattern>
                   <rawString>true</rawString>
                 </relocation>
+                <relocation>
+                  <pattern>META-INF/versions/(\d+)/org/roaringbitmap/</pattern>
+                  <shadedPattern>META-INF/versions/$1/org/apache/pulsar/shade/org/roaringbitmap/</shadedPattern>
+                  <rawString>true</rawString>
+                </relocation>
                 <relocation>
                   <pattern>META-INF/versions/(\d+)/org/yaml/</pattern>
                   <shadedPattern>META-INF/versions/$1/org/apache/pulsar/shade/org/yaml/</shadedPattern>
@@ -439,6 +461,10 @@
                   <pattern>org.reactivestreams</pattern>
                   <shadedPattern>org.apache.pulsar.shade.org.reactivestreams</shadedPattern>
                 </relocation>
+                <relocation>
+                  <pattern>org.roaringbitmap</pattern>
+                  <shadedPattern>org.apache.pulsar.shade.org.roaringbitmap</shadedPattern>
+                </relocation>
                 <relocation>
                   <pattern>org.tukaani</pattern>
                   <shadedPattern>org.apache.pulsar.shade.org.tukaani</shadedPattern>

pulsar-client-dependencies-minimized/pom.xml

@@ -0,0 +1,100 @@
+<?xml version="1.0"?>
+<!--
+
+    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.
+
+-->
+<project
+        xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"
+        xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
+  <modelVersion>4.0.0</modelVersion>
+  <parent>
+    <groupId>org.apache.pulsar</groupId>
+    <artifactId>pulsar</artifactId>
+    <version>4.1.0-SNAPSHOT</version>
+  </parent>
+
+  <artifactId>pulsar-client-dependencies-minimized</artifactId>
+  <name>Apache Pulsar :: Client :: Dependencies minimized</name>
+  <description>This module is used in `pulsar-client-all`, `pulsar-client-shaded`, and `pulsar-client-admin-shaded`
+    to minimize the number of classes included in the shaded jars for specific dependencies.
+    Currently, it is used to minimize the classes included from `fastutil`.
+  </description>
+  <dependencies>
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>pulsar-client-original</artifactId>
+      <version>${project.version}</version>
+    </dependency>
+  </dependencies>
+  <build>
+    <finalName>${project.artifactId}-${project.version}</finalName>
+    <plugins>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-deploy-plugin</artifactId>
+        <!-- Skips the deployment of the minimized dependencies to Maven Central as this is an intermediate
+ module used for building the shaded client jars -->
+        <configuration>
+          <skip>true</skip>
+        </configuration>
+      </plugin>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-shade-plugin</artifactId>
+        <executions>
+          <execution>
+            <phase>package</phase>
+            <goals>
+              <goal>shade</goal>
+            </goals>
+            <configuration>
+              <createDependencyReducedPom>true</createDependencyReducedPom>
+              <promoteTransitiveDependencies>false</promoteTransitiveDependencies>
+              <!-- minimize the classes included in the shaded jar -->
+              <minimizeJar>true</minimizeJar>
+              <artifactSet>
+                <includes>
+                  <!-- The Pulsar module that references the library being minimized -->
+                  <include>org.apache.pulsar:pulsar-client-original</include>
+                  <!-- Currently, only fastutil is minimized -->
+                  <include>it.unimi.dsi:fastutil</include>
+                </includes>
+              </artifactSet>
+              <filters>
+                <!--
+                This filter specifies the classes that use the dependencies.
+                Both includes and excludes are set to **.
+                -->
+                <filter>
+                  <artifact>org.apache.pulsar:pulsar-client-original</artifact>
+                  <includes>
+                    <include>**</include>
+                  </includes>
+                  <excludes>
+                    <exclude>**</exclude>
+                  </excludes>
+                </filter>
+              </filters>
+            </configuration>
+          </execution>
+        </executions>
+      </plugin>
+    </plugins>
+  </build>
+</project>

pulsar-client-shaded/pom.xml

@@ -39,6 +39,17 @@
       <groupId>${project.groupId}</groupId>
       <artifactId>pulsar-client-original</artifactId>
       <version>${project.version}</version>
+      <exclusions>
+        <exclusion>
+          <groupId>it.unimi.dsi</groupId>
+          <artifactId>fastutil</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>pulsar-client-dependencies-minimized</artifactId>
+      <version>${project.version}</version>
     </dependency>
     <dependency>
       <groupId>${project.groupId}</groupId>
@@ -164,6 +175,8 @@
                   <include>org.reactivestreams:reactive-streams</include>
                   <include>org.tukaani:xz</include>
                   <include>org.yaml:snakeyaml</include>
+                  <include>org.apache.pulsar:pulsar-client-dependencies-minimized</include>
+                  <include>org.roaringbitmap:RoaringBitmap</include>
                 </includes>
                 <excludes>
                   <exclude>com.fasterxml.jackson.core:jackson-annotations</exclude>
@@ -263,6 +276,10 @@
                   <pattern>io.swagger</pattern>
                   <shadedPattern>org.apache.pulsar.shade.io.swagger</shadedPattern>
                 </relocation>
+                <relocation>
+                  <pattern>it.unimi.dsi.fastutil</pattern>
+                  <shadedPattern>org.apache.pulsar.shade.it.unimi.dsi.fastutil</shadedPattern>
+                </relocation>
                 <relocation>
                   <pattern>javax.activation</pattern>
                   <shadedPattern>org.apache.pulsar.shade.javax.activation</shadedPattern>
@@ -281,6 +298,11 @@
                   </shadedPattern>
                   <rawString>true</rawString>
                 </relocation>
+                <relocation>
+                  <pattern>META-INF/versions/(\d+)/org/roaringbitmap/</pattern>
+                  <shadedPattern>META-INF/versions/$1/org/apache/pulsar/shade/org/roaringbitmap/</shadedPattern>
+                  <rawString>true</rawString>
+                </relocation>
                 <relocation>
                   <pattern>META-INF/versions/(\d+)/org/yaml/</pattern>
                   <shadedPattern>META-INF/versions/$1/org/apache/pulsar/shade/org/yaml/</shadedPattern>
@@ -343,6 +365,10 @@
                   <pattern>org.reactivestreams</pattern>
                   <shadedPattern>org.apache.pulsar.shade.org.reactivestreams</shadedPattern>
                 </relocation>
+                <relocation>
+                  <pattern>org.roaringbitmap</pattern>
+                  <shadedPattern>org.apache.pulsar.shade.org.roaringbitmap</shadedPattern>
+                </relocation>
                 <relocation>
                   <pattern>org.tukaani</pattern>
                   <shadedPattern>org.apache.pulsar.shade.org.tukaani</shadedPattern>