[branch-3.3] Bump pulsar 3.3.0.2 (#1261)

streamnative · Jun 12, 2024 · 985d291 · 985d291
1 parent 0d6d81d
commit 985d291
Show file tree

Hide file tree

Showing 5 changed files with 27 additions and 24 deletions.
diff --git a/amqp-impl/src/main/java/io/streamnative/pulsar/handlers/amqp/AmqpEntryWriter.java b/amqp-impl/src/main/java/io/streamnative/pulsar/handlers/amqp/AmqpEntryWriter.java
@@ -97,7 +97,7 @@ public void addComplete(Position position, ByteBuf entryData, Object ctx) {
             log.debug("[{}] Success to write entry with position {}.", topic.getName(), position);
         }
         topic.recordAddLatency(System.nanoTime() - context.startTimeNs, TimeUnit.MICROSECONDS);
-        topic.getTransactionBuffer().syncMaxReadPositionForNormalPublish((PositionImpl) position);
+        topic.getTransactionBuffer().syncMaxReadPositionForNormalPublish((PositionImpl) position, false);
         context.positionFuture.complete(position);
         context.recycle();
     }

diff --git a/...l/src/test/java/io/streamnative/pulsar/handlers/amqp/test/AopProtocolHandlerTestBase.java b/...l/src/test/java/io/streamnative/pulsar/handlers/amqp/test/AopProtocolHandlerTestBase.java
@@ -27,6 +27,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
 import java.util.function.Supplier;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
@@ -204,21 +205,21 @@ public void reallyShutdown() {
     private final BookKeeperClientFactory mockBookKeeperClientFactory = new BookKeeperClientFactory() {
 
         @Override
-        public BookKeeper create(ServiceConfiguration conf, MetadataStoreExtended store,
-                                 EventLoopGroup eventLoopGroup,
-                                 Optional<Class<? extends EnsemblePlacementPolicy>> ensemblePlacementPolicyClass,
-                                 Map<String, Object> properties) {
+        public CompletableFuture<BookKeeper> create(ServiceConfiguration conf, MetadataStoreExtended store,
+                                        EventLoopGroup eventLoopGroup,
+                                        Optional<Class<? extends EnsemblePlacementPolicy>> ensemblePlacementPolicyClass,
+                                        Map<String, Object> properties) {
             // Always return the same instance (so that we don't loose the mock BK content on broker restart
-            return mockBookKeeper;
+            return CompletableFuture.completedFuture(mockBookKeeper);
         }
 
         @Override
-        public BookKeeper create(ServiceConfiguration conf, MetadataStoreExtended store,
+        public CompletableFuture<BookKeeper> create(ServiceConfiguration conf, MetadataStoreExtended store,
                                  EventLoopGroup eventLoopGroup,
                                  Optional<Class<? extends EnsemblePlacementPolicy>> ensemblePlacementPolicyClass,
                                  Map<String, Object> properties, StatsLogger statsLogger) {
             // Always return the same instance (so that we don't loose the mock BK content on broker restart
-            return mockBookKeeper;
+            return CompletableFuture.completedFuture(mockBookKeeper);
         }
 
         @Override

diff --git a/pom.xml b/pom.xml
@@ -40,7 +40,7 @@
     <project.compiler.release>${maven.compiler.target}</project.compiler.release>
 
     <!-- dependencies -->
-    <pulsar.version>3.3.0-SNAPSHOT</pulsar.version>
+    <pulsar.version>3.3.0.2</pulsar.version>
     <qpid-protocol-plugin.version>8.0.0</qpid-protocol-plugin.version>
     <rabbitmq.version>5.8.0</rabbitmq.version>
 
@@ -386,7 +386,7 @@
     </repository>
     <repository>
       <id>ossrh</id>
-      <url>https://s01.oss.sonatype.org/service/local/repositories/iostreamnative-3095/content</url>
+      <url>https://s01.oss.sonatype.org/service/local/repositories/iostreamnative-3155/content</url>
     </repository>
   </repositories>
 

diff --git a/...lient/src/test/java/io/streamnative/pulsar/handlers/amqp/AmqpProtocolHandlerTestBase.java b/...lient/src/test/java/io/streamnative/pulsar/handlers/amqp/AmqpProtocolHandlerTestBase.java
@@ -52,6 +52,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Predicate;
@@ -332,21 +333,21 @@ public void reallyShutdown() {
     private final BookKeeperClientFactory mockBookKeeperClientFactory = new BookKeeperClientFactory() {
 
         @Override
-        public BookKeeper create(ServiceConfiguration conf, MetadataStoreExtended store,
-                                 EventLoopGroup eventLoopGroup,
-                                 Optional<Class<? extends EnsemblePlacementPolicy>> ensemblePlacementPolicyClass,
-                                 Map<String, Object> properties) {
+        public CompletableFuture<BookKeeper> create(ServiceConfiguration conf, MetadataStoreExtended store,
+                                        EventLoopGroup eventLoopGroup,
+                                        Optional<Class<? extends EnsemblePlacementPolicy>> ensemblePlacementPolicyClass,
+                                        Map<String, Object> properties) {
             // Always return the same instance (so that we don't loose the mock BK content on broker restart
-            return mockBookKeeper;
+            return CompletableFuture.completedFuture(mockBookKeeper);
         }
 
         @Override
-        public BookKeeper create(ServiceConfiguration conf, MetadataStoreExtended store,
+        public CompletableFuture<BookKeeper> create(ServiceConfiguration conf, MetadataStoreExtended store,
                                  EventLoopGroup eventLoopGroup,
                                  Optional<Class<? extends EnsemblePlacementPolicy>> ensemblePlacementPolicyClass,
                                  Map<String, Object> properties, StatsLogger statsLogger) {
             // Always return the same instance (so that we don't loose the mock BK content on broker restart
-            return mockBookKeeper;
+            return CompletableFuture.completedFuture(mockBookKeeper);
         }
 
         @Override

diff --git a/tests/src/test/java/io/streamnative/pulsar/handlers/amqp/AmqpProtocolHandlerTestBase.java b/tests/src/test/java/io/streamnative/pulsar/handlers/amqp/AmqpProtocolHandlerTestBase.java
@@ -30,6 +30,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Predicate;
@@ -359,21 +360,21 @@ public void reallyShutdown() {
     private final BookKeeperClientFactory mockBookKeeperClientFactory = new BookKeeperClientFactory() {
 
         @Override
-        public BookKeeper create(ServiceConfiguration conf, MetadataStoreExtended store,
-                                 EventLoopGroup eventLoopGroup,
-                                 Optional<Class<? extends EnsemblePlacementPolicy>> ensemblePlacementPolicyClass,
-                                 Map<String, Object> properties) {
+        public CompletableFuture<BookKeeper> create(ServiceConfiguration conf, MetadataStoreExtended store,
+                                        EventLoopGroup eventLoopGroup,
+                                        Optional<Class<? extends EnsemblePlacementPolicy>> ensemblePlacementPolicyClass,
+                                        Map<String, Object> properties) {
             // Always return the same instance (so that we don't loose the mock BK content on broker restart
-            return mockBookKeeper;
+            return CompletableFuture.completedFuture(mockBookKeeper);
         }
 
         @Override
-        public BookKeeper create(ServiceConfiguration conf, MetadataStoreExtended store,
+        public CompletableFuture<BookKeeper> create(ServiceConfiguration conf, MetadataStoreExtended store,
                                  EventLoopGroup eventLoopGroup,
                                  Optional<Class<? extends EnsemblePlacementPolicy>> ensemblePlacementPolicyClass,
                                  Map<String, Object> properties, StatsLogger statsLogger) {
             // Always return the same instance (so that we don't loose the mock BK content on broker restart
-            return mockBookKeeper;
+            return CompletableFuture.completedFuture(mockBookKeeper);
         }
 
         @Override