This is an automated email from the ASF dual-hosted git repository.
merlimat pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 12d8aa9a977 [fix][test] Run makeReadEntryProbFail's errorOrNot on a
caller-provided executor (#26123)
12d8aa9a977 is described below
commit 12d8aa9a9773f2546f58aca4678c65f6d506dd95
Author: Lari Hotari <[email protected]>
AuthorDate: Wed Jul 1 06:32:42 2026 +0300
[fix][test] Run makeReadEntryProbFail's errorOrNot on a caller-provided
executor (#26123)
---
.../bookkeeper/mledger/impl/ManagedLedgerTest.java | 21 ++++++++++++---------
.../pulsar/broker/service/OneWayReplicatorTest.java | 4 +++-
.../PersistentReplicatorInflightTaskTest.java | 21 ++++++++++++++++++---
3 files changed, 33 insertions(+), 13 deletions(-)
diff --git
a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
index 67d2ab17e4f..f186d5ca176 100644
---
a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
+++
b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
@@ -77,6 +77,7 @@ import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.FutureTask;
@@ -188,19 +189,21 @@ public class ManagedLedgerTest extends
MockedBookKeeperTestCase {
ml.currentLedger = spyLedgerHandle;
}
- public static void makeReadEntryProbFail(ManagedLedgerImpl ml,
Supplier<ManagedLedgerException> errorOrNot)
- throws Exception {
+ public static void makeReadEntryProbFail(ManagedLedgerImpl ml,
Supplier<ManagedLedgerException> errorOrNot,
+ Executor errorSupplierExecutor)
throws Exception {
ml.entryCache.clear();
LedgerHandle currentLedger = ml.currentLedger;
final LedgerHandle spyLedgerHandle = spy(currentLedger);
doAnswer(invocation -> {
- long ledgerId = (long) invocation.getArguments()[0];
- long entryId = (long) invocation.getArguments()[1];
- ManagedLedgerException mightError = errorOrNot.get();
- if (mightError != null) {
- return CompletableFuture.failedFuture(mightError);
- }
- return currentLedger.readUnconfirmedAsync(ledgerId, entryId);
+ long ledgerId = invocation.getArgument(0);
+ long entryId = invocation.getArgument(1);
+ // Evaluate errorOrNot on errorSupplierExecutor. Pass a
single-threaded executor when errorOrNot may
+ // block (e.g. it waits on a CountDownLatch) so it doesn't block
the calling read thread; pass
+ // MoreExecutors.directExecutor() to evaluate it inline on the
calling thread.
+ return CompletableFuture.supplyAsync(errorOrNot,
errorSupplierExecutor)
+ .thenCompose(mightError -> mightError != null
+ ?
CompletableFuture.<LedgerEntries>failedFuture(mightError)
+ : currentLedger.readUnconfirmedAsync(ledgerId,
entryId));
}).when(spyLedgerHandle).readUnconfirmedAsync(anyLong(), anyLong());
ml.currentLedger = spyLedgerHandle;
}
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java
index 7f69b485509..f4f24de6b2d 100644
---
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java
@@ -36,6 +36,7 @@ import com.fasterxml.jackson.databind.ObjectMapper;
import com.github.benmanes.caffeine.cache.AsyncLoadingCache;
import com.google.common.collect.Multimap;
import com.google.common.collect.Sets;
+import com.google.common.util.concurrent.MoreExecutors;
import io.netty.channel.Channel;
import io.netty.util.concurrent.FastThreadLocalThread;
import java.lang.reflect.Field;
@@ -937,7 +938,8 @@ public class OneWayReplicatorTest extends
OneWayReplicatorTestBase {
}
return new ManagedLedgerException.TooManyRequestsException("mocked
error");
};
- ManagedLedgerTest.makeReadEntryProbFail(ml1, bkErrorOrNot);
+ // bkErrorOrNot doesn't block, so evaluate it inline on the calling
read thread via directExecutor().
+ ManagedLedgerTest.makeReadEntryProbFail(ml1, bkErrorOrNot,
MoreExecutors.directExecutor());
// Verify: the replication will finish even though received
ManagedLedgerException.TooManyRequestsException.
pulsar1.getConfig().setReplicationStartAt("earliest");
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentReplicatorInflightTaskTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentReplicatorInflightTaskTest.java
index db1a49d7340..e790b97a0d4 100644
---
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentReplicatorInflightTaskTest.java
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentReplicatorInflightTaskTest.java
@@ -46,9 +46,12 @@ import java.util.List;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
+import lombok.Cleanup;
import lombok.CustomLog;
import org.apache.bookkeeper.mledger.Entry;
import org.apache.bookkeeper.mledger.ManagedCursor;
@@ -162,6 +165,10 @@ public class PersistentReplicatorInflightTaskTest extends
OneWayReplicatorTestBa
PersistentTopic topic = (PersistentTopic)
pulsar1.getBrokerService().getTopic(topicName, false)
.join().get();
ManagedLedgerImpl ml = (ManagedLedgerImpl)
topic.getManagedLedger();
+ // errorOrNot blocks on failRead, so run it on a dedicated
single-threaded executor instead of the
+ // calling read thread; otherwise the blocked read thread would
stall replicator.terminate().
+ @Cleanup("shutdownNow")
+ ExecutorService readFailExecutor =
Executors.newSingleThreadExecutor();
ManagedLedgerTest.makeReadEntryProbFail(ml, () -> {
readStarted.countDown();
try {
@@ -173,7 +180,7 @@ public class PersistentReplicatorInflightTaskTest extends
OneWayReplicatorTestBa
return new ManagedLedgerException(e);
}
return new
ManagedLedgerException.TooManyRequestsException("mocked read failure");
- });
+ }, readFailExecutor);
pulsar1.getConfig().setReplicationStartAt("earliest");
admin1.topics().setReplicationClusters(topicName,
Arrays.asList(cluster1, cluster2));
@@ -324,6 +331,10 @@ public class PersistentReplicatorInflightTaskTest extends
OneWayReplicatorTestBa
ManagedLedgerImpl ml = (ManagedLedgerImpl)
topic.getManagedLedger();
// Clear the entry cache and intercept ledger reads: block the
first read so it stays in flight
// (the in-flight task keeps entries == null), then let it and all
later reads succeed.
+ // errorOrNot blocks the first read on releaseRead, so run it on a
dedicated single-threaded executor
+ // instead of the calling read thread.
+ @Cleanup("shutdownNow")
+ ExecutorService readFailExecutor =
Executors.newSingleThreadExecutor();
ManagedLedgerTest.makeReadEntryProbFail(ml, () -> {
if (blockNextRead.compareAndSet(true, false)) {
readBlocked.countDown();
@@ -335,7 +346,7 @@ public class PersistentReplicatorInflightTaskTest extends
OneWayReplicatorTestBa
}
// Never fail the read; it must complete successfully to reach
the skip branch.
return null;
- });
+ }, readFailExecutor);
// Start replication from earliest; the first read blocks inside
the ledger read (in flight).
pulsar1.getConfig().setReplicationStartAt("earliest");
@@ -467,6 +478,10 @@ public class PersistentReplicatorInflightTaskTest extends
OneWayReplicatorTestBa
ManagedLedgerImpl ml = (ManagedLedgerImpl)
topic.getManagedLedger();
// Clear the entry cache and intercept ledger reads: block the
first read so it stays in flight
// (the in-flight task keeps entries == null), then let it and all
later reads succeed.
+ // errorOrNot blocks the first read on releaseRead, so run it on a
dedicated single-threaded executor
+ // instead of the calling read thread.
+ @Cleanup("shutdownNow")
+ ExecutorService readFailExecutor =
Executors.newSingleThreadExecutor();
ManagedLedgerTest.makeReadEntryProbFail(ml, () -> {
if (blockNextRead.compareAndSet(true, false)) {
readBlocked.countDown();
@@ -478,7 +493,7 @@ public class PersistentReplicatorInflightTaskTest extends
OneWayReplicatorTestBa
}
// Never fail the read; it must complete successfully to reach
the skip branch.
return null;
- });
+ }, readFailExecutor);
// Start replication from earliest; the first read blocks inside
the ledger read (in flight).
pulsar1.getConfig().setReplicationStartAt("earliest");
