[improve][broker]Improve ManagedLedger search position by offset

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedger.java

@@ -711,6 +711,17 @@ default void skipNonRecoverableLedger(long ledgerId){}
      * */
     CompletableFuture<Position> asyncFindPosition(Predicate<Entry> predicate);
 
+    /**
+     * Optimized find position by offset with predicate.
+     * Uses first entry index metadata to skip ledgers where the target offset
+     * is smaller than the ledger's first entry index, improving performance.
+     *
+     * @param offset the target offset to find
+     * @param predicate the predicate to test entries
+     * @return CompletableFuture<Position> the position where the predicate matches
+     */
+    CompletableFuture<Position> asyncFindPosition(long offset, Predicate<Entry> predicate);
+
     /**
      * Get the ManagedLedgerInterceptor for ManagedLedger.
      * */

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/intercept/ManagedLedgerInterceptor.java

@@ -75,6 +75,7 @@ default void afterFailedAddEntry(int numberOfMessages){
      */
     void onManagedLedgerPropertiesInitialize(Map<String, String> propertiesMap);
 
+    long getIndex();
     /**
      * A handle for reading the last ledger entry.
      */

pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/ManagedLedgerInterceptorImpl.java

@@ -71,6 +71,7 @@ public ManagedLedgerInterceptorImpl(Set<BrokerEntryMetadataInterceptor> brokerEn
         }
     }
 
+    @Override
     public long getIndex() {
         long index = -1;
 

pulsar-broker/src/test/java/org/apache/bookkeeper/mledger/impl/CustomizedManagedLedgerStorageForTest.java

@@ -610,6 +610,11 @@ public CompletableFuture<Position> asyncFindPosition(Predicate<Entry> predicate)
             return delegate.asyncFindPosition(predicate);
         }
 
+        @Override
+        public CompletableFuture<Position> asyncFindPosition(long offset, Predicate<Entry> predicate) {
+            return delegate.asyncFindPosition(offset, predicate);
+        }
+
         @Override
         public ManagedLedgerInterceptor getManagedLedgerInterceptor() {
             return delegate.getManagedLedgerInterceptor();

pulsar-transaction/coordinator/src/main/java/org/apache/pulsar/transaction/coordinator/impl/MLTransactionSequenceIdGenerator.java

@@ -90,4 +90,9 @@ long generateSequenceId() {
     long getCurrentSequenceId() {
         return sequenceId.get();
     }
+
+    @Override
+    public long getIndex() {
+        return -1L;
+    }
 }

tiered-storage/jcloud/src/test/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/MockManagedLedger.java

@@ -352,6 +352,11 @@ public CompletableFuture<Position> asyncFindPosition(Predicate<Entry> predicate)
         return CompletableFuture.completedFuture(null);
     }
 
+    @Override
+    public CompletableFuture<Position> asyncFindPosition(long offset, Predicate<Entry> predicate) {
+        return CompletableFuture.completedFuture(null);
+    }
+
     @Override
     public ManagedLedgerInterceptor getManagedLedgerInterceptor() {
         return null;