Skip to content

Commit d377bc9

Browse files
thetumbledlhotari
thetumbled
and
lhotari
authored
[improve][client] PIP-393: Improve performance of Negative Acknowledgement (#23600)
Co-authored-by: Lari Hotari <[email protected]>
1 parent fd45029 commit d377bc9

File tree

11 files changed

+344
-51
lines changed

11 files changed

+344
-51
lines changed

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

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -418,6 +418,8 @@ The Apache Software License, Version 2.0
418418
- avro-protobuf-1.11.4.jar
419419
* RE2j -- re2j-1.7.jar
420420
* Spotify completable-futures -- completable-futures-0.3.6.jar
421+
* RoaringBitmap -- RoaringBitmap-1.2.0.jar
422+
* Fastutil -- fastutil-8.5.14.jar
421423

422424
BSD 3-clause "New" or "Revised" License
423425
* JSR305 -- jsr305-3.0.2.jar -- ../licenses/LICENSE-JSR305.txt

‎pom.xml‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2580,6 +2580,7 @@ flexible messaging model and an intuitive client API.</description>
25802580
<module>pulsar-metadata</module>
25812581
<module>jetcd-core-shaded</module>
25822582
<module>jclouds-shaded</module>
2583+
<module>pulsar-client-dependencies-minimized</module>
25832584

25842585
<!-- package management releated modules (begin) -->
25852586
<module>pulsar-package-management</module>
@@ -2645,6 +2646,7 @@ flexible messaging model and an intuitive client API.</description>
26452646
<module>distribution</module>
26462647
<module>pulsar-metadata</module>
26472648
<module>jetcd-core-shaded</module>
2649+
<module>pulsar-client-dependencies-minimized</module>
26482650
<!-- package management releated modules (begin) -->
26492651
<module>pulsar-package-management</module>
26502652
<!-- package management releated modules (end) -->

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

Lines changed: 48 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
package org.apache.pulsar.client.impl;
2020

2121
import static org.testng.Assert.assertEquals;
22+
import static org.testng.Assert.assertNotNull;
2223
import static org.testng.Assert.assertNull;
2324
import static org.testng.Assert.assertTrue;
2425
import java.util.HashSet;
@@ -311,19 +312,64 @@ public void testNegativeAcksDeleteFromUnackedTracker() throws Exception {
311312
// negative topic message id
312313
consumer.negativeAcknowledge(topicMessageId);
313314
NegativeAcksTracker negativeAcksTracker = consumer.getNegativeAcksTracker();
314-
assertEquals(negativeAcksTracker.getNackedMessagesCount().orElse((long) -1).longValue(), 1L);
315+
assertEquals(negativeAcksTracker.getNackedMessagesCount(), 1L);
315316
assertEquals(unAckedMessageTracker.size(), 0);
316317
negativeAcksTracker.close();
317318
// negative batch message id
318319
unAckedMessageTracker.add(messageId);
319320
consumer.negativeAcknowledge(batchMessageId);
320321
consumer.negativeAcknowledge(batchMessageId2);
321322
consumer.negativeAcknowledge(batchMessageId3);
322-
assertEquals(negativeAcksTracker.getNackedMessagesCount().orElse((long) -1).longValue(), 1L);
323+
assertEquals(negativeAcksTracker.getNackedMessagesCount(), 1L);
323324
assertEquals(unAckedMessageTracker.size(), 0);
324325
negativeAcksTracker.close();
325326
}
326327

328+
/**
329+
* If we nack multiple messages in the same batch with different redelivery delays, the messages should be redelivered
330+
* with the correct delay. However, all messages are redelivered at the same time.
331+
* @throws Exception
332+
*/
333+
@Test
334+
public void testNegativeAcksWithBatch() throws Exception {
335+
cleanup();
336+
conf.setAcknowledgmentAtBatchIndexLevelEnabled(true);
337+
setup();
338+
String topic = BrokerTestUtil.newUniqueName("testNegativeAcksWithBatch");
339+
340+
@Cleanup
341+
Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
342+
.topic(topic)
343+
.subscriptionName("sub1")
344+
.acknowledgmentGroupTime(0, TimeUnit.SECONDS)
345+
.subscriptionType(SubscriptionType.Shared)
346+
.enableBatchIndexAcknowledgment(true)
347+
.negativeAckRedeliveryDelay(3, TimeUnit.SECONDS)
348+
.subscribe();
349+
350+
@Cleanup
351+
Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
352+
.topic(topic)
353+
.enableBatching(true)
354+
.batchingMaxPublishDelay(1, TimeUnit.HOURS)
355+
.batchingMaxMessages(2)
356+
.create();
357+
// send two messages in the same batch
358+
producer.sendAsync("test-0");
359+
producer.sendAsync("test-1");
360+
producer.flush();
361+
362+
// negative ack the first message
363+
consumer.negativeAcknowledge(consumer.receive());
364+
// wait for 2s, negative ack the second message
365+
Thread.sleep(2000);
366+
consumer.negativeAcknowledge(consumer.receive());
367+
368+
// now 2s has passed, the first message should be redelivered 1s later.
369+
Message<String> msg1 = consumer.receive(2, TimeUnit.SECONDS);
370+
assertNotNull(msg1);
371+
}
372+
327373
@Test
328374
public void testNegativeAcksWithBatchAckEnabled() throws Exception {
329375
cleanup();

‎pulsar-client-admin-shaded/pom.xml‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,17 @@
3434
<groupId>${project.groupId}</groupId>
3535
<artifactId>pulsar-client-admin-original</artifactId>
3636
<version>${project.version}</version>
37+
<exclusions>
38+
<exclusion>
39+
<groupId>it.unimi.dsi</groupId>
40+
<artifactId>fastutil</artifactId>
41+
</exclusion>
42+
</exclusions>
43+
</dependency>
44+
<dependency>
45+
<groupId>${project.groupId}</groupId>
46+
<artifactId>pulsar-client-dependencies-minimized</artifactId>
47+
<version>${project.version}</version>
3748
</dependency>
3849
<dependency>
3950
<groupId>${project.groupId}</groupId>
@@ -150,6 +161,8 @@
150161
<include>org.objenesis:*</include>
151162
<include>org.reactivestreams:reactive-streams</include>
152163
<include>org.yaml:snakeyaml</include>
164+
<include>org.apache.pulsar:pulsar-client-dependencies-minimized</include>
165+
<include>org.roaringbitmap:RoaringBitmap</include>
153166
</includes>
154167
<excludes>
155168
<exclude>com.fasterxml.jackson.core:jackson-annotations</exclude>
@@ -269,6 +282,10 @@
269282
<pattern>io.swagger</pattern>
270283
<shadedPattern>org.apache.pulsar.shade.io.swagger</shadedPattern>
271284
</relocation>
285+
<relocation>
286+
<pattern>it.unimi.dsi.fastutil</pattern>
287+
<shadedPattern>org.apache.pulsar.shade.it.unimi.dsi.fastutil</shadedPattern>
288+
</relocation>
272289
<relocation>
273290
<pattern>javassist</pattern>
274291
<shadedPattern>org.apache.pulsar.shade.javassist</shadedPattern>
@@ -313,6 +330,11 @@
313330
<shadedPattern>META-INF/versions/$1/org/apache/pulsar/shade/org/glassfish/</shadedPattern>
314331
<rawString>true</rawString>
315332
</relocation>
333+
<relocation>
334+
<pattern>META-INF/versions/(\d+)/org/roaringbitmap/</pattern>
335+
<shadedPattern>META-INF/versions/$1/org/apache/pulsar/shade/org/roaringbitmap/</shadedPattern>
336+
<rawString>true</rawString>
337+
</relocation>
316338
<relocation>
317339
<pattern>META-INF/versions/(\d+)/org/yaml/</pattern>
318340
<shadedPattern>META-INF/versions/$1/org/apache/pulsar/shade/org/yaml/</shadedPattern>
@@ -374,6 +396,10 @@
374396
<pattern>org.reactivestreams</pattern>
375397
<shadedPattern>org.apache.pulsar.shade.org.reactivestreams</shadedPattern>
376398
</relocation>
399+
<relocation>
400+
<pattern>org.roaringbitmap</pattern>
401+
<shadedPattern>org.apache.pulsar.shade.org.roaringbitmap</shadedPattern>
402+
</relocation>
377403
<relocation>
378404
<pattern>org.yaml</pattern>
379405
<shadedPattern>org.apache.pulsar.shade.org.yaml</shadedPattern>

‎pulsar-client-all/pom.xml‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,17 @@
3939
<groupId>${project.groupId}</groupId>
4040
<artifactId>pulsar-client-original</artifactId>
4141
<version>${project.version}</version>
42+
<exclusions>
43+
<exclusion>
44+
<groupId>it.unimi.dsi</groupId>
45+
<artifactId>fastutil</artifactId>
46+
</exclusion>
47+
</exclusions>
48+
</dependency>
49+
<dependency>
50+
<groupId>${project.groupId}</groupId>
51+
<artifactId>pulsar-client-dependencies-minimized</artifactId>
52+
<version>${project.version}</version>
4253
</dependency>
4354
<dependency>
4455
<groupId>${project.groupId}</groupId>
@@ -200,6 +211,8 @@
200211
<include>org.reactivestreams:reactive-streams</include>
201212
<include>org.tukaani:xz</include>
202213
<include>org.yaml:snakeyaml</include>
214+
<include>org.apache.pulsar:pulsar-client-dependencies-minimized</include>
215+
<include>org.roaringbitmap:RoaringBitmap</include>
203216
</includes>
204217
<excludes>
205218
<exclude>com.fasterxml.jackson.core:jackson-annotations</exclude>
@@ -317,6 +330,10 @@
317330
<pattern>io.swagger</pattern>
318331
<shadedPattern>org.apache.pulsar.shade.io.swagger</shadedPattern>
319332
</relocation>
333+
<relocation>
334+
<pattern>it.unimi.dsi.fastutil</pattern>
335+
<shadedPattern>org.apache.pulsar.shade.it.unimi.dsi.fastutil</shadedPattern>
336+
</relocation>
320337
<relocation>
321338
<pattern>javassist</pattern>
322339
<shadedPattern>org.apache.pulsar.shade.javassist</shadedPattern>
@@ -361,6 +378,11 @@
361378
<shadedPattern>META-INF/versions/$1/org/apache/pulsar/shade/org/glassfish/</shadedPattern>
362379
<rawString>true</rawString>
363380
</relocation>
381+
<relocation>
382+
<pattern>META-INF/versions/(\d+)/org/roaringbitmap/</pattern>
383+
<shadedPattern>META-INF/versions/$1/org/apache/pulsar/shade/org/roaringbitmap/</shadedPattern>
384+
<rawString>true</rawString>
385+
</relocation>
364386
<relocation>
365387
<pattern>META-INF/versions/(\d+)/org/yaml/</pattern>
366388
<shadedPattern>META-INF/versions/$1/org/apache/pulsar/shade/org/yaml/</shadedPattern>
@@ -439,6 +461,10 @@
439461
<pattern>org.reactivestreams</pattern>
440462
<shadedPattern>org.apache.pulsar.shade.org.reactivestreams</shadedPattern>
441463
</relocation>
464+
<relocation>
465+
<pattern>org.roaringbitmap</pattern>
466+
<shadedPattern>org.apache.pulsar.shade.org.roaringbitmap</shadedPattern>
467+
</relocation>
442468
<relocation>
443469
<pattern>org.tukaani</pattern>
444470
<shadedPattern>org.apache.pulsar.shade.org.tukaani</shadedPattern>

‎pulsar-client-dependencies-minimized/pom.xml‎

Lines changed: 100 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,100 @@
1+
<?xml version="1.0"?>
2+
<!--
3+
4+
Licensed to the Apache Software Foundation (ASF) under one
5+
or more contributor license agreements. See the NOTICE file
6+
distributed with this work for additional information
7+
regarding copyright ownership. The ASF licenses this file
8+
to you under the Apache License, Version 2.0 (the
9+
"License"); you may not use this file except in compliance
10+
with the License. You may obtain a copy of the License at
11+
12+
http://www.apache.org/licenses/LICENSE-2.0
13+
14+
Unless required by applicable law or agreed to in writing,
15+
software distributed under the License is distributed on an
16+
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
17+
KIND, either express or implied. See the License for the
18+
specific language governing permissions and limitations
19+
under the License.
20+
21+
-->
22+
<project
23+
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"
24+
xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
25+
<modelVersion>4.0.0</modelVersion>
26+
<parent>
27+
<groupId>org.apache.pulsar</groupId>
28+
<artifactId>pulsar</artifactId>
29+
<version>4.1.0-SNAPSHOT</version>
30+
</parent>
31+
32+
<artifactId>pulsar-client-dependencies-minimized</artifactId>
33+
<name>Apache Pulsar :: Client :: Dependencies minimized</name>
34+
<description>This module is used in `pulsar-client-all`, `pulsar-client-shaded`, and `pulsar-client-admin-shaded`
35+
to minimize the number of classes included in the shaded jars for specific dependencies.
36+
Currently, it is used to minimize the classes included from `fastutil`.
37+
</description>
38+
<dependencies>
39+
<dependency>
40+
<groupId>${project.groupId}</groupId>
41+
<artifactId>pulsar-client-original</artifactId>
42+
<version>${project.version}</version>
43+
</dependency>
44+
</dependencies>
45+
<build>
46+
<finalName>${project.artifactId}-${project.version}</finalName>
47+
<plugins>
48+
<plugin>
49+
<groupId>org.apache.maven.plugins</groupId>
50+
<artifactId>maven-deploy-plugin</artifactId>
51+
<!-- Skips the deployment of the minimized dependencies to Maven Central as this is an intermediate
52+
module used for building the shaded client jars -->
53+
<configuration>
54+
<skip>true</skip>
55+
</configuration>
56+
</plugin>
57+
<plugin>
58+
<groupId>org.apache.maven.plugins</groupId>
59+
<artifactId>maven-shade-plugin</artifactId>
60+
<executions>
61+
<execution>
62+
<phase>package</phase>
63+
<goals>
64+
<goal>shade</goal>
65+
</goals>
66+
<configuration>
67+
<createDependencyReducedPom>true</createDependencyReducedPom>
68+
<promoteTransitiveDependencies>false</promoteTransitiveDependencies>
69+
<!-- minimize the classes included in the shaded jar -->
70+
<minimizeJar>true</minimizeJar>
71+
<artifactSet>
72+
<includes>
73+
<!-- The Pulsar module that references the library being minimized -->
74+
<include>org.apache.pulsar:pulsar-client-original</include>
75+
<!-- Currently, only fastutil is minimized -->
76+
<include>it.unimi.dsi:fastutil</include>
77+
</includes>
78+
</artifactSet>
79+
<filters>
80+
<!--
81+
This filter specifies the classes that use the dependencies.
82+
Both includes and excludes are set to **.
83+
-->
84+
<filter>
85+
<artifact>org.apache.pulsar:pulsar-client-original</artifact>
86+
<includes>
87+
<include>**</include>
88+
</includes>
89+
<excludes>
90+
<exclude>**</exclude>
91+
</excludes>
92+
</filter>
93+
</filters>
94+
</configuration>
95+
</execution>
96+
</executions>
97+
</plugin>
98+
</plugins>
99+
</build>
100+
</project>

‎pulsar-client-shaded/pom.xml‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,17 @@
3939
<groupId>${project.groupId}</groupId>
4040
<artifactId>pulsar-client-original</artifactId>
4141
<version>${project.version}</version>
42+
<exclusions>
43+
<exclusion>
44+
<groupId>it.unimi.dsi</groupId>
45+
<artifactId>fastutil</artifactId>
46+
</exclusion>
47+
</exclusions>
48+
</dependency>
49+
<dependency>
50+
<groupId>${project.groupId}</groupId>
51+
<artifactId>pulsar-client-dependencies-minimized</artifactId>
52+
<version>${project.version}</version>
4253
</dependency>
4354
<dependency>
4455
<groupId>${project.groupId}</groupId>
@@ -164,6 +175,8 @@
164175
<include>org.reactivestreams:reactive-streams</include>
165176
<include>org.tukaani:xz</include>
166177
<include>org.yaml:snakeyaml</include>
178+
<include>org.apache.pulsar:pulsar-client-dependencies-minimized</include>
179+
<include>org.roaringbitmap:RoaringBitmap</include>
167180
</includes>
168181
<excludes>
169182
<exclude>com.fasterxml.jackson.core:jackson-annotations</exclude>
@@ -263,6 +276,10 @@
263276
<pattern>io.swagger</pattern>
264277
<shadedPattern>org.apache.pulsar.shade.io.swagger</shadedPattern>
265278
</relocation>
279+
<relocation>
280+
<pattern>it.unimi.dsi.fastutil</pattern>
281+
<shadedPattern>org.apache.pulsar.shade.it.unimi.dsi.fastutil</shadedPattern>
282+
</relocation>
266283
<relocation>
267284
<pattern>javax.activation</pattern>
268285
<shadedPattern>org.apache.pulsar.shade.javax.activation</shadedPattern>
@@ -281,6 +298,11 @@
281298
</shadedPattern>
282299
<rawString>true</rawString>
283300
</relocation>
301+
<relocation>
302+
<pattern>META-INF/versions/(\d+)/org/roaringbitmap/</pattern>
303+
<shadedPattern>META-INF/versions/$1/org/apache/pulsar/shade/org/roaringbitmap/</shadedPattern>
304+
<rawString>true</rawString>
305+
</relocation>
284306
<relocation>
285307
<pattern>META-INF/versions/(\d+)/org/yaml/</pattern>
286308
<shadedPattern>META-INF/versions/$1/org/apache/pulsar/shade/org/yaml/</shadedPattern>
@@ -343,6 +365,10 @@
343365
<pattern>org.reactivestreams</pattern>
344366
<shadedPattern>org.apache.pulsar.shade.org.reactivestreams</shadedPattern>
345367
</relocation>
368+
<relocation>
369+
<pattern>org.roaringbitmap</pattern>
370+
<shadedPattern>org.apache.pulsar.shade.org.roaringbitmap</shadedPattern>
371+
</relocation>
346372
<relocation>
347373
<pattern>org.tukaani</pattern>
348374
<shadedPattern>org.apache.pulsar.shade.org.tukaani</shadedPattern>

0 commit comments

Comments
 (0)