[branch-3.0][fix] Change read max position to earliest position (#1329)

streamnative · Aug 23, 2024 · afbc11e · afbc11e
1 parent db0a68a
commit afbc11e
Show file tree

Hide file tree

Showing 3 changed files with 6 additions and 4 deletions.
diff --git a/amqp-impl/src/main/java/io/streamnative/pulsar/handlers/amqp/AmqpExchangeReplicator.java b/amqp-impl/src/main/java/io/streamnative/pulsar/handlers/amqp/AmqpExchangeReplicator.java
@@ -32,6 +32,7 @@
 import org.apache.bookkeeper.mledger.ManagedCursor;
 import org.apache.bookkeeper.mledger.ManagedLedgerException;
 import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
 import org.apache.commons.collections4.CollectionUtils;
 import org.apache.commons.lang3.tuple.Pair;
 import org.apache.pulsar.broker.service.persistent.PersistentTopic;
@@ -193,7 +194,7 @@ private void readMoreEntries() {
                 if (log.isDebugEnabled()) {
                     log.debug("{} Schedule read of {} messages.", name, messagesToRead);
                 }
-                cursor.asyncReadEntriesOrWait(messagesToRead, readMaxSizeBytes, this, null, null);
+                cursor.asyncReadEntriesOrWait(messagesToRead, readMaxSizeBytes, this, null, PositionImpl.LATEST);
             } else {
                 if (log.isDebugEnabled()) {
                     log.debug("{} Not schedule read due to pending read. Messages to read {}.",

diff --git a/amqp-impl/src/main/java/io/streamnative/pulsar/handlers/amqp/ExchangeMessageRouter.java b/amqp-impl/src/main/java/io/streamnative/pulsar/handlers/amqp/ExchangeMessageRouter.java
@@ -46,6 +46,7 @@
 import org.apache.bookkeeper.mledger.Position;
 import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
 import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
 import org.apache.pulsar.broker.PulsarServerException;
 import org.apache.pulsar.broker.service.persistent.PersistentTopic;
 import org.apache.pulsar.client.api.MessageId;
@@ -161,7 +162,7 @@ public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
                         HAVE_PENDING_READ_UPDATER.set(ExchangeMessageRouter.this, FALSE);
                         log.error("Failed to read entries from exchange {}", exchange.getName(), exception);
                     }
-                }, null, null);
+                }, null, PositionImpl.LATEST);
             } else {
                 log.warn("{} Not schedule read due to pending read. Messages to read {}.",
                         exchange.getName(), availablePermits);

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.0.5.5</pulsar.version>
+    <pulsar.version>3.0.6.3</pulsar.version>
     <qpid-protocol-plugin.version>8.0.0</qpid-protocol-plugin.version>
     <rabbitmq.version>5.8.0</rabbitmq.version>
 
@@ -387,7 +387,7 @@
 
     <repository>
       <id>ossrh</id>
-      <url>https://s01.oss.sonatype.org/service/local/repositories/iostreamnative-3192/content</url>
+      <url>https://s01.oss.sonatype.org/service/local/repositories/iostreamnative-3277/content</url>
     </repository>
   </repositories>