You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/05/18 06:19:21 UTC
(pulsar) branch branch-4.2 updated (eee5d9d3ec1 -> 1042e4f818f)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from eee5d9d3ec1 [improve][build] Upgrade org.apache.kerby:kerb-simplekdc from 1.1.1 to 2.1.1 (#25785)
new 18dd778f18b [refactor][fn] Use Map instead of TreeMap for connector/function API types (#25790)
new 1042e4f818f [improve][broker] Prevent stale replicator pending reads after termination (#25767)
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../service/persistent/PersistentReplicator.java | 21 ++++++--
.../PersistentReplicatorInflightTaskTest.java | 60 ++++++++++++++++++++++
.../org/apache/pulsar/functions/LocalRunner.java | 11 ++--
.../pulsar/functions/worker/ConnectorsManager.java | 7 +--
.../pulsar/functions/worker/FunctionsManager.java | 9 ++--
.../functions/utils/functions/FunctionUtils.java | 3 +-
.../pulsar/functions/utils/io/ConnectorUtils.java | 4 +-
.../functions/utils/io/ReloadConnectorsResult.java | 4 +-
.../utils/io/ConnectorUtilsReloadTest.java | 14 ++---
9 files changed, 104 insertions(+), 29 deletions(-)
(pulsar) 01/02: [refactor][fn] Use Map instead of TreeMap for connector/function API types (#25790)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 18dd778f18b6da50bcb6427d3fbbdca263c5043d
Author: Dream95 <86...@qq.com>
AuthorDate: Sat May 16 16:58:59 2026 +0800
[refactor][fn] Use Map instead of TreeMap for connector/function API types (#25790)
Signed-off-by: Dream95 <zh...@163.com>
(cherry picked from commit eba873d9c3926b806f5462af834867ccb692fdfb)
---
.../main/java/org/apache/pulsar/functions/LocalRunner.java | 11 +++++------
.../apache/pulsar/functions/worker/ConnectorsManager.java | 7 ++++---
.../apache/pulsar/functions/worker/FunctionsManager.java | 9 +++++----
.../pulsar/functions/utils/functions/FunctionUtils.java | 3 ++-
.../apache/pulsar/functions/utils/io/ConnectorUtils.java | 4 ++--
.../pulsar/functions/utils/io/ReloadConnectorsResult.java | 4 ++--
.../functions/utils/io/ConnectorUtilsReloadTest.java | 14 +++++++-------
7 files changed, 27 insertions(+), 25 deletions(-)
diff --git a/pulsar-functions/localrun/src/main/java/org/apache/pulsar/functions/LocalRunner.java b/pulsar-functions/localrun/src/main/java/org/apache/pulsar/functions/LocalRunner.java
index 1e2d4e0b21b..bfb3eab6ff3 100644
--- a/pulsar-functions/localrun/src/main/java/org/apache/pulsar/functions/LocalRunner.java
+++ b/pulsar-functions/localrun/src/main/java/org/apache/pulsar/functions/LocalRunner.java
@@ -38,7 +38,6 @@ import java.util.Map;
import java.util.Optional;
import java.util.Timer;
import java.util.TimerTask;
-import java.util.TreeMap;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
@@ -724,7 +723,7 @@ public class LocalRunner implements AutoCloseable {
private ClassLoader isBuiltInFunction(String functionType) throws IOException {
// Validate the connector type from the locally available connectors
- TreeMap<String, FunctionArchive> functions = getFunctions();
+ Map<String, FunctionArchive> functions = getFunctions();
String functionName = functionType.replaceFirst("^builtin://", "");
FunctionArchive function = functions.get(functionName);
@@ -738,7 +737,7 @@ public class LocalRunner implements AutoCloseable {
private ClassLoader isBuiltInSource(String sourceType) throws IOException {
// Validate the connector type from the locally available connectors
- TreeMap<String, Connector> connectors = getConnectors();
+ Map<String, Connector> connectors = getConnectors();
String source = sourceType.replaceFirst("^builtin://", "");
Connector connector = connectors.get(source);
@@ -752,7 +751,7 @@ public class LocalRunner implements AutoCloseable {
private ClassLoader isBuiltInSink(String sinkType) throws IOException {
// Validate the connector type from the locally available connectors
- TreeMap<String, Connector> connectors = getConnectors();
+ Map<String, Connector> connectors = getConnectors();
String sink = sinkType.replaceFirst("^builtin://", "");
Connector connector = connectors.get(sink);
@@ -764,11 +763,11 @@ public class LocalRunner implements AutoCloseable {
}
}
- private TreeMap<String, FunctionArchive> getFunctions() throws IOException {
+ private Map<String, FunctionArchive> getFunctions() throws IOException {
return FunctionUtils.searchForFunctions(functionsDir, narExtractionDirectory, true);
}
- private TreeMap<String, Connector> getConnectors() throws IOException {
+ private Map<String, Connector> getConnectors() throws IOException {
return ConnectorUtils.searchForConnectors(connectorsDir, narExtractionDirectory, true);
}
diff --git a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/worker/ConnectorsManager.java b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/worker/ConnectorsManager.java
index a8ee6f3ce20..74435f99bcf 100644
--- a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/worker/ConnectorsManager.java
+++ b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/worker/ConnectorsManager.java
@@ -23,6 +23,7 @@ import java.io.IOException;
import java.nio.file.Path;
import java.util.Collection;
import java.util.List;
+import java.util.Map;
import java.util.TreeMap;
import java.util.stream.Collectors;
import lombok.Getter;
@@ -38,7 +39,7 @@ import org.apache.pulsar.functions.utils.io.ReloadConnectorsResult;
public class ConnectorsManager implements AutoCloseable {
@Getter
- private volatile TreeMap<String, Connector> connectors;
+ private volatile Map<String, Connector> connectors;
@VisibleForTesting
public ConnectorsManager() {
@@ -49,7 +50,7 @@ public class ConnectorsManager implements AutoCloseable {
this.connectors = createConnectors(workerConfig);
}
- private static TreeMap<String, Connector> createConnectors(WorkerConfig workerConfig) throws IOException {
+ private static Map<String, Connector> createConnectors(WorkerConfig workerConfig) throws IOException {
boolean enableClassloading = isEnableClassloading(workerConfig);
return ConnectorUtils.searchForConnectors(workerConfig.getConnectorsDirectory(),
workerConfig.getNarExtractionDirectory(), enableClassloading);
@@ -119,7 +120,7 @@ public class ConnectorsManager implements AutoCloseable {
});
}
- private void closeConnectors(TreeMap<String, Connector> connectorMap) {
+ private void closeConnectors(Map<String, Connector> connectorMap) {
closeConnectors(connectorMap.values());
connectorMap.clear();
}
diff --git a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/worker/FunctionsManager.java b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/worker/FunctionsManager.java
index 5ab7ff7221a..cdd77249502 100644
--- a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/worker/FunctionsManager.java
+++ b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/worker/FunctionsManager.java
@@ -22,6 +22,7 @@ import com.google.common.annotations.VisibleForTesting;
import java.io.IOException;
import java.nio.file.Path;
import java.util.List;
+import java.util.Map;
import java.util.TreeMap;
import java.util.stream.Collectors;
import lombok.extern.slf4j.Slf4j;
@@ -32,7 +33,7 @@ import org.apache.pulsar.functions.utils.functions.FunctionUtils;
@Slf4j
public class FunctionsManager implements AutoCloseable {
- private TreeMap<String, FunctionArchive> functions;
+ private Map<String, FunctionArchive> functions;
@VisibleForTesting
public FunctionsManager() {
@@ -61,12 +62,12 @@ public class FunctionsManager implements AutoCloseable {
}
public void reloadFunctions(WorkerConfig workerConfig) throws IOException {
- TreeMap<String, FunctionArchive> oldFunctions = functions;
+ Map<String, FunctionArchive> oldFunctions = functions;
this.functions = createFunctions(workerConfig);
closeFunctions(oldFunctions);
}
- private static TreeMap<String, FunctionArchive> createFunctions(WorkerConfig workerConfig) throws IOException {
+ private static Map<String, FunctionArchive> createFunctions(WorkerConfig workerConfig) throws IOException {
boolean enableClassloading = workerConfig.getEnableClassloadingOfBuiltinFiles()
|| ThreadRuntimeFactory.class.getName().equals(workerConfig.getFunctionRuntimeFactoryClassName());
return FunctionUtils.searchForFunctions(workerConfig.getFunctionsDirectory(),
@@ -79,7 +80,7 @@ public class FunctionsManager implements AutoCloseable {
closeFunctions(functions);
}
- private void closeFunctions(TreeMap<String, FunctionArchive> functionMap) {
+ private void closeFunctions(Map<String, FunctionArchive> functionMap) {
functionMap.values().forEach(functionArchive -> {
try {
functionArchive.close();
diff --git a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/functions/FunctionUtils.java b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/functions/FunctionUtils.java
index 24f801f3da2..f4d45edf363 100644
--- a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/functions/FunctionUtils.java
+++ b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/functions/FunctionUtils.java
@@ -25,6 +25,7 @@ import java.nio.file.DirectoryStream;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
+import java.util.Map;
import java.util.TreeMap;
import lombok.experimental.UtilityClass;
import lombok.extern.slf4j.Slf4j;
@@ -75,7 +76,7 @@ public class FunctionUtils {
.readValue(narClassLoader.getServiceDefinition(PULSAR_IO_SERVICE_NAME), valueType);
}
- public static TreeMap<String, FunctionArchive> searchForFunctions(String functionsDirectory,
+ public static Map<String, FunctionArchive> searchForFunctions(String functionsDirectory,
String narExtractionDirectory,
boolean enableClassloading) throws IOException {
Path path = Paths.get(functionsDirectory).toAbsolutePath().normalize();
diff --git a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/io/ConnectorUtils.java b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/io/ConnectorUtils.java
index d08a4942884..0ff7e9427b3 100644
--- a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/io/ConnectorUtils.java
+++ b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/io/ConnectorUtils.java
@@ -165,7 +165,7 @@ public class ConnectorUtils {
return fields;
}
- public static TreeMap<String, Connector> searchForConnectors(String connectorsDirectory,
+ public static Map<String, Connector> searchForConnectors(String connectorsDirectory,
String narExtractionDirectory,
boolean enableClassloading) throws IOException {
Path path = Paths.get(connectorsDirectory).toAbsolutePath().normalize();
@@ -210,7 +210,7 @@ public class ConnectorUtils {
* connectors the caller should close
*/
public static ReloadConnectorsResult reloadConnectors(
- TreeMap<String, Connector> previous,
+ Map<String, Connector> previous,
String connectorsDirectory,
String narExtractionDirectory,
boolean enableClassloading) throws IOException {
diff --git a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/io/ReloadConnectorsResult.java b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/io/ReloadConnectorsResult.java
index cdfab35692c..b46aa38fead 100644
--- a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/io/ReloadConnectorsResult.java
+++ b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/io/ReloadConnectorsResult.java
@@ -19,11 +19,11 @@
package org.apache.pulsar.functions.utils.io;
import java.util.List;
-import java.util.TreeMap;
+import java.util.Map;
/**
* Result of {@link ConnectorUtils#reloadConnectors}: the new connector map and connectors evicted from the active set
* that the caller must close.
*/
-public record ReloadConnectorsResult(TreeMap<String, Connector> connectors, List<Connector> connectorsToClose) {
+public record ReloadConnectorsResult(Map<String, Connector> connectors, List<Connector> connectorsToClose) {
}
diff --git a/pulsar-functions/utils/src/test/java/org/apache/pulsar/functions/utils/io/ConnectorUtilsReloadTest.java b/pulsar-functions/utils/src/test/java/org/apache/pulsar/functions/utils/io/ConnectorUtilsReloadTest.java
index 07b6459369a..0e05912d8c9 100644
--- a/pulsar-functions/utils/src/test/java/org/apache/pulsar/functions/utils/io/ConnectorUtilsReloadTest.java
+++ b/pulsar-functions/utils/src/test/java/org/apache/pulsar/functions/utils/io/ConnectorUtilsReloadTest.java
@@ -27,7 +27,7 @@ import java.io.IOException;
import java.io.OutputStream;
import java.nio.file.Files;
import java.nio.file.Path;
-import java.util.TreeMap;
+import java.util.Map;
import java.util.zip.ZipEntry;
import java.util.zip.ZipOutputStream;
import org.apache.pulsar.common.io.ConnectorDefinition;
@@ -77,7 +77,7 @@ public class ConnectorUtilsReloadTest {
Path nar = dir.resolve("c1.nar");
writeMinimalNar(nar, sampleDefinition("c-one"));
- TreeMap<String, Connector> first =
+ Map<String, Connector> first =
ConnectorUtils.searchForConnectors(dir.toString(), NarClassLoader.DEFAULT_NAR_EXTRACTION_DIR, false);
Connector c1 = first.get("c-one");
c1.getConnectorFunctionPackage();
@@ -86,7 +86,7 @@ public class ConnectorUtilsReloadTest {
first, dir.toString(), NarClassLoader.DEFAULT_NAR_EXTRACTION_DIR, false);
assertTrue(reload.connectorsToClose().isEmpty());
closeEvicted(reload);
- TreeMap<String, Connector> second = reload.connectors();
+ Map<String, Connector> second = reload.connectors();
assertSame(second.get("c-one"), c1);
c1.getConnectorFunctionPackage();
@@ -98,7 +98,7 @@ public class ConnectorUtilsReloadTest {
Path nar = dir.resolve("c1.nar");
writeMinimalNar(nar, sampleDefinition("c-one"));
- TreeMap<String, Connector> first =
+ Map<String, Connector> first =
ConnectorUtils.searchForConnectors(dir.toString(), NarClassLoader.DEFAULT_NAR_EXTRACTION_DIR, false);
Connector before = first.get("c-one");
@@ -109,7 +109,7 @@ public class ConnectorUtilsReloadTest {
ReloadConnectorsResult reload = ConnectorUtils.reloadConnectors(
first, dir.toString(), NarClassLoader.DEFAULT_NAR_EXTRACTION_DIR, false);
closeEvicted(reload);
- TreeMap<String, Connector> second = reload.connectors();
+ Map<String, Connector> second = reload.connectors();
assertNotSame(second.get("c-one"), before);
assertThrows(IllegalStateException.class, before::getConnectorFunctionPackage);
@@ -123,7 +123,7 @@ public class ConnectorUtilsReloadTest {
writeMinimalNar(nar1, sampleDefinition("conn-a"));
writeMinimalNar(nar2, sampleDefinition("conn-b"));
- TreeMap<String, Connector> first =
+ Map<String, Connector> first =
ConnectorUtils.searchForConnectors(dir.toString(), NarClassLoader.DEFAULT_NAR_EXTRACTION_DIR, false);
Connector removed = first.get("conn-b");
Files.delete(nar2);
@@ -131,7 +131,7 @@ public class ConnectorUtilsReloadTest {
ReloadConnectorsResult reload = ConnectorUtils.reloadConnectors(
first, dir.toString(), NarClassLoader.DEFAULT_NAR_EXTRACTION_DIR, false);
closeEvicted(reload);
- TreeMap<String, Connector> second = reload.connectors();
+ Map<String, Connector> second = reload.connectors();
assertEquals(second.size(), 1);
assertSame(second.get("conn-a"), first.get("conn-a"));
(pulsar) 02/02: [improve][broker] Prevent stale replicator pending reads after termination (#25767)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 1042e4f818ff41d6e4f3d64b3811fa0027f06435
Author: sinan liu <li...@gmail.com>
AuthorDate: Mon May 18 09:49:50 2026 +0800
[improve][broker] Prevent stale replicator pending reads after termination (#25767)
(cherry picked from commit 8f9f5b49d631e235e86d79e48a63722e74db4413)
---
.../service/persistent/PersistentReplicator.java | 21 ++++++--
.../PersistentReplicatorInflightTaskTest.java | 60 ++++++++++++++++++++++
2 files changed, 77 insertions(+), 4 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java
index 4202f812a4a..b125bd0306b 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java
@@ -501,7 +501,7 @@ public abstract class PersistentReplicator extends AbstractReplicator
@Override
public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
- InFlightTask inFlightTask = (InFlightTask) ctx;
+ completeFailedReadTask(ctx);
if (state != Started) {
log.info("[{}] Replicator was disconnected while reading entries."
+ " Stop reading. Replicator state: {}",
@@ -522,11 +522,9 @@ public abstract class PersistentReplicator extends AbstractReplicator
terminate();
return;
} else if (!(exception instanceof TooManyRequestsException)) {
- inFlightTask.setEntries(Collections.emptyList());
log.error("[{}] Error reading entries at {}. Retrying to read in {}s. ({})",
replicatorId, ctx, waitTimeMillis / 1000.0, exception.getMessage(), exception);
} else {
- inFlightTask.setEntries(Collections.emptyList());
log.debug("[{}] Throttled by bookies while reading at {}. Retrying to read in {}s. ({})",
replicatorId, ctx, waitTimeMillis / 1000.0, exception.getMessage(),
exception);
@@ -535,6 +533,21 @@ public abstract class PersistentReplicator extends AbstractReplicator
brokerService.executor().schedule(this::readMoreEntries, waitTimeMillis, TimeUnit.MILLISECONDS);
}
+ private void completeFailedReadTask(Object ctx) {
+ if (!(ctx instanceof InFlightTask)) {
+ log.error("[{}] Unexpected read entries failed context {}", replicatorId, ctx);
+ return;
+ }
+
+ InFlightTask inFlightTask = (InFlightTask) ctx;
+ if (inFlightTask.entries == null) {
+ inFlightTask.setEntries(Collections.emptyList());
+ } else {
+ log.error("[{}] Unexpected completed in-flight task in read entries failed callback. inFlightTask={}",
+ replicatorId, inFlightTask);
+ }
+ }
+
public CompletableFuture<Void> clearBacklog() {
CompletableFuture<Void> future = new CompletableFuture<>();
@@ -990,4 +1003,4 @@ public abstract class PersistentReplicator extends AbstractReplicator
String getReplicatorId() {
return replicatorId;
}
-}
\ No newline at end of file
+}
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 e8a7a11a2db..c825d49caa6 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
@@ -28,17 +28,25 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.mledger.Entry;
+import org.apache.bookkeeper.mledger.ManagedLedgerException;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.PositionFactory;
+import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
+import org.apache.bookkeeper.mledger.impl.ManagedLedgerTest;
import org.apache.pulsar.broker.BrokerTestUtil;
import org.apache.pulsar.broker.service.AbstractReplicator;
import org.apache.pulsar.broker.service.BrokerServiceInternalMethodInvoker;
import org.apache.pulsar.broker.service.OneWayReplicatorTestBase;
import org.apache.pulsar.broker.service.persistent.PersistentReplicator.InFlightTask;
import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.Schema;
+import org.awaitility.Awaitility;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.testng.Assert;
@@ -107,6 +115,58 @@ public class PersistentReplicatorInflightTaskTest extends OneWayReplicatorTestBa
assertTrue(counter.get() <= 1);
}
+ @Test
+ public void testReadEntriesFailedCompletesInFlightTaskAfterReplicatorTerminated() throws Exception {
+ String topicName = BrokerTestUtil.newUniqueName("persistent://" + nonReplicatedNamespace + "/tp_");
+ CountDownLatch readStarted = new CountDownLatch(1);
+ CountDownLatch failRead = new CountDownLatch(1);
+ Producer<String> producer = null;
+ try {
+ admin1.topics().createNonPartitionedTopic(topicName);
+ admin2.topics().createNonPartitionedTopic(topicName);
+ producer = client1.newProducer(Schema.STRING).topic(topicName).create();
+ producer.send("msg");
+
+ PersistentTopic topic = (PersistentTopic) pulsar1.getBrokerService().getTopic(topicName, false)
+ .join().get();
+ ManagedLedgerImpl ml = (ManagedLedgerImpl) topic.getManagedLedger();
+ ManagedLedgerTest.makeReadEntryProbFail(ml, () -> {
+ readStarted.countDown();
+ try {
+ if (!failRead.await(30, TimeUnit.SECONDS)) {
+ return new ManagedLedgerException("Timed out waiting to fail read entries");
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return new ManagedLedgerException(e);
+ }
+ return new ManagedLedgerException.TooManyRequestsException("mocked read failure");
+ });
+
+ pulsar1.getConfig().setReplicationStartAt("earliest");
+ admin1.topics().setReplicationClusters(topicName, Arrays.asList(cluster1, cluster2));
+ assertTrue(readStarted.await(30, TimeUnit.SECONDS));
+
+ PersistentReplicator replicator = (PersistentReplicator) topic.getReplicators().get(cluster2);
+ Assert.assertNotNull(replicator, "Replicator should not be null");
+ Assert.assertTrue(replicator.hasPendingRead());
+
+ replicator.terminate();
+ Assert.assertTrue(replicator.getState() != AbstractReplicator.State.Started);
+ failRead.countDown();
+
+ Awaitility.await().atMost(30, TimeUnit.SECONDS).untilAsserted(() ->
+ Assert.assertFalse(replicator.hasPendingRead()));
+ } finally {
+ failRead.countDown();
+ if (producer != null) {
+ producer.close();
+ }
+ admin1.topics().delete(topicName, true);
+ admin2.topics().delete(topicName, true);
+ }
+ }
+
@Test
public void testCreateOrRecycleInFlightTaskIntoQueue() throws Exception {
log.info("Starting testCreateOrRecycleInFlightTaskIntoQueue");