You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by mm...@apache.org on 2026/09/17 16:05:40 UTC
(pulsar) branch master updated: [fix][test] Avoid unsafe cleanup between shared-cluster tests (#26611)
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 136cdf6ef04b [fix][test] Avoid unsafe cleanup between shared-cluster tests (#26611)
136cdf6ef04b is described below
commit 136cdf6ef04bd345d963c26dbf6a4277c5c7aeeb
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Thu Sep 17 19:05:34 2026 +0300
[fix][test] Avoid unsafe cleanup between shared-cluster tests (#26611)
---
.../main/kotlin/pulsar.java-conventions.gradle.kts | 2 -
buildtools/build.gradle.kts | 2 +-
.../tests/FastThreadLocalCleanupListener.java | 64 -----
.../pulsar/tests/FastThreadLocalStateCleaner.java | 138 ----------
.../pulsar/tests/MockitoCleanupListener.java | 64 -----
.../tests/FastThreadLocalStateCleanerTest.java | 128 ---------
.../apache/pulsar/broker/service/ServerCnx.java | 28 +-
.../broker/service/persistent/PersistentTopic.java | 3 +-
.../service/GetLastMessageIdEntryLeakTest.java | 102 ++++---
...tentDispatcherMultipleConsumersClassicTest.java | 84 ++----
.../PersistentDispatcherMultipleConsumersTest.java | 112 ++++----
...rsistentDispatcherSingleActiveConsumerTest.java | 186 ++++++-------
.../service/persistent/ReaderMessageTTLTest.java | 19 +-
.../SharedDispatcherPermitAccountingTest.java | 299 +++++++++++++--------
.../pulsar/broker/stats/SubscriptionStatsTest.java | 9 +-
.../pulsar/client/api/MultiTopicsConsumerTest.java | 43 ++-
.../client/impl/ProducerReconnectionTest.java | 31 ++-
.../client/impl/SimpleProduceConsumeIoTest.java | 39 +--
.../stats/client/PulsarBrokerStatsClientTest.java | 3 +-
.../common/util/netty/ChannelFuturesTest.java | 24 +-
.../websocket/admin/WebSocketWebResourceTest.java | 37 ++-
21 files changed, 504 insertions(+), 913 deletions(-)
diff --git a/build-logic/conventions/src/main/kotlin/pulsar.java-conventions.gradle.kts b/build-logic/conventions/src/main/kotlin/pulsar.java-conventions.gradle.kts
index 5ec724a8bb90..885be271cf81 100644
--- a/build-logic/conventions/src/main/kotlin/pulsar.java-conventions.gradle.kts
+++ b/build-logic/conventions/src/main/kotlin/pulsar.java-conventions.gradle.kts
@@ -223,8 +223,6 @@ tasks.withType<Test>().configureEach {
"org.apache.pulsar.tests.PulsarTestListener",
"org.apache.pulsar.tests.AnnotationListener",
"org.apache.pulsar.tests.FailFastNotifier",
- "org.apache.pulsar.tests.MockitoCleanupListener",
- "org.apache.pulsar.tests.FastThreadLocalCleanupListener",
"org.apache.pulsar.tests.ThreadLeakDetectorListener",
"org.apache.pulsar.tests.SingletonCleanerListener",
))
diff --git a/buildtools/build.gradle.kts b/buildtools/build.gradle.kts
index af4dc417f66f..7ec0e69f3285 100644
--- a/buildtools/build.gradle.kts
+++ b/buildtools/build.gradle.kts
@@ -41,7 +41,7 @@ dependencies {
implementation(libs.jcl.over.slf4j)
implementation(libs.commons.lang3)
- // Netty is needed at runtime for ExtendedNettyLeakDetector and FastThreadLocalCleanupListener.
+ // Netty is needed at runtime for ExtendedNettyLeakDetector.
// Using runtimeOnly so it propagates transitively when other modules depend on buildtools.
compileOnly(libs.netty.common)
runtimeOnly(libs.netty.common)
diff --git a/buildtools/src/main/java/org/apache/pulsar/tests/FastThreadLocalCleanupListener.java b/buildtools/src/main/java/org/apache/pulsar/tests/FastThreadLocalCleanupListener.java
deleted file mode 100644
index 46ee5d057c46..000000000000
--- a/buildtools/src/main/java/org/apache/pulsar/tests/FastThreadLocalCleanupListener.java
+++ /dev/null
@@ -1,64 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.pulsar.tests;
-
-import java.util.List;
-import lombok.CustomLog;
-import org.testng.ITestClass;
-
-/**
- * Cleanup Thread Local state attach to Netty's FastThreadLocal.
- */
-@CustomLog
-public class FastThreadLocalCleanupListener extends BetweenTestClassesListenerAdapter {
- private static final boolean FAST_THREAD_LOCAL_CLEANUP_ENABLED =
- Boolean.parseBoolean(System.getProperty("testFastThreadLocalCleanup", "true"));
- private static final String FAST_THREAD_LOCAL_CLEANUP_PACKAGE =
- System.getProperty("testFastThreadLocalCleanupPackage", "org.apache.pulsar");
- private static final FastThreadLocalStateCleaner CLEANER = new FastThreadLocalStateCleaner(object -> {
- if ("*".equals(FAST_THREAD_LOCAL_CLEANUP_PACKAGE)) {
- return true;
- }
- Class<?> clazz = object.getClass();
- if (clazz.isArray()) {
- clazz = clazz.getComponentType();
- }
- Package pkg = clazz.getPackage();
- if (pkg != null && pkg.getName() != null) {
- return pkg.getName()
- .startsWith(FAST_THREAD_LOCAL_CLEANUP_PACKAGE);
- } else {
- return false;
- }
- });
-
- @Override
- protected void onBetweenTestClasses(List<ITestClass> testClasses) {
- if (FAST_THREAD_LOCAL_CLEANUP_ENABLED && FastThreadLocalStateCleaner.isEnabled()) {
- log.info("Cleaning up FastThreadLocal thread local state.");
- CLEANER.cleanupAllFastThreadLocals((thread, value) -> {
- log.info().attr("thread", thread)
- .attr("class", value.getClass().getName())
- .attr("value", value)
- .log("Cleaning FastThreadLocal state");
- });
- }
- }
-
-}
diff --git a/buildtools/src/main/java/org/apache/pulsar/tests/FastThreadLocalStateCleaner.java b/buildtools/src/main/java/org/apache/pulsar/tests/FastThreadLocalStateCleaner.java
deleted file mode 100644
index 6b1a954d97be..000000000000
--- a/buildtools/src/main/java/org/apache/pulsar/tests/FastThreadLocalStateCleaner.java
+++ /dev/null
@@ -1,138 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.pulsar.tests;
-
-import java.lang.reflect.Field;
-import java.lang.reflect.InvocationTargetException;
-import java.util.Objects;
-import java.util.function.BiConsumer;
-import java.util.function.Predicate;
-import lombok.CustomLog;
-import org.apache.commons.lang3.ClassUtils;
-import org.apache.commons.lang3.ThreadUtils;
-import org.apache.commons.lang3.reflect.FieldUtils;
-
-/**
- * Cleanup Thread Local state attach to Netty's FastThreadLocal.
- *
- * This is not thread-safe, but that aspect is ignored.
- */
-@CustomLog
-public final class FastThreadLocalStateCleaner {
- private static final ThreadLocal<?> SLOW_THREAD_LOCAL_MAP = lookupSlowThreadLocalMap();
- private static final Class<?> FAST_THREAD_LOCAL_CLASS;
- private static final Field THREAD_LOCAL_MAP_FIELD;
- private static final Field INDEXED_VARIABLES_FIELD;
- private static final Object UNSET_OBJECT;
-
- static {
- Class<?> clazz = null;
- Field threadLocalMapField = null;
- Field indexedVariablesField = null;
- Object unsetObject = null;
- if (SLOW_THREAD_LOCAL_MAP != null) {
- try {
- clazz = ClassUtils.getClass("io.netty.util.concurrent.FastThreadLocalThread");
- Class<?> internalThreadLocalMapClass =
- ClassUtils.getClass("io.netty.util.internal.InternalThreadLocalMap");
- threadLocalMapField = FieldUtils
- .getDeclaredField(clazz, "threadLocalMap", true);
- indexedVariablesField = FieldUtils.getDeclaredField(internalThreadLocalMapClass,
- "indexedVariables", true);
- Field unsetField = FieldUtils.getField(internalThreadLocalMapClass, "UNSET");
- unsetObject = unsetField.get(null);
- } catch (ClassNotFoundException | IllegalAccessException e) {
- // ignore
- log.debug().exception(e).log("Ignoring exception");
- clazz = null;
- threadLocalMapField = null;
- indexedVariablesField = null;
- unsetObject = null;
- }
- }
- FAST_THREAD_LOCAL_CLASS = clazz;
- THREAD_LOCAL_MAP_FIELD = threadLocalMapField;
- INDEXED_VARIABLES_FIELD = indexedVariablesField;
- UNSET_OBJECT = unsetObject;
- }
-
- private final Predicate<Object> valueFilter;
-
- private static ThreadLocal<?> lookupSlowThreadLocalMap() {
- try {
- Field slowThreadLocalMapField = FieldUtils.getDeclaredField(
- ClassUtils.getClass("io.netty.util.internal.InternalThreadLocalMap"),
- "slowThreadLocalMap", true);
- if (slowThreadLocalMapField != null) {
- return (ThreadLocal<?>) slowThreadLocalMapField.get(null);
- } else {
- log.warn("Cannot find InternalThreadLocalMap.slowThreadLocalMap field."
- + " This might be due to using an unsupported netty-common version.");
- return null;
- }
- } catch (IllegalAccessException | ClassNotFoundException e) {
- log.warn().exception(e).log("Cannot find InternalThreadLocalMap.slowThreadLocalMap thread local");
- return null;
- }
- }
-
- public FastThreadLocalStateCleaner(Predicate<Object> valueFilter) {
- this.valueFilter = valueFilter;
- }
-
- public void cleanupAllFastThreadLocals(Thread thread, BiConsumer<Thread, Object> cleanedValueListener) {
- Objects.nonNull(thread);
- try {
- Object internalThreadLocalMap;
- if (FAST_THREAD_LOCAL_CLASS.isInstance(thread)) {
- internalThreadLocalMap = THREAD_LOCAL_MAP_FIELD.get(thread);
- } else {
- internalThreadLocalMap = ThreadLocalStateCleaner.INSTANCE
- .getThreadLocalValue(SLOW_THREAD_LOCAL_MAP, thread);
- }
- if (internalThreadLocalMap != null) {
- Object[] indexedVariables = (Object[]) INDEXED_VARIABLES_FIELD.get(internalThreadLocalMap);
- for (int i = 0; i < indexedVariables.length; i++) {
- Object threadLocalValue = indexedVariables[i];
- if (threadLocalValue != UNSET_OBJECT && threadLocalValue != null) {
- if (valueFilter.test(threadLocalValue)) {
- indexedVariables[i] = UNSET_OBJECT;
- if (cleanedValueListener != null) {
- cleanedValueListener.accept(thread, threadLocalValue);
- }
- }
- }
- }
- }
- } catch (IllegalAccessException | InvocationTargetException e) {
- log.warn().attr("thread", thread).exception(e).log("Cannot reset state for FastLocalThread");
- }
- }
-
- // cleanup all fast thread local state on all active threads
- public void cleanupAllFastThreadLocals(BiConsumer<Thread, Object> cleanedValueListener) {
- for (Thread thread : ThreadUtils.getAllThreads()) {
- cleanupAllFastThreadLocals(thread, cleanedValueListener);
- }
- }
-
- public static boolean isEnabled() {
- return SLOW_THREAD_LOCAL_MAP != null && FAST_THREAD_LOCAL_CLASS != null;
- }
-}
diff --git a/buildtools/src/main/java/org/apache/pulsar/tests/MockitoCleanupListener.java b/buildtools/src/main/java/org/apache/pulsar/tests/MockitoCleanupListener.java
deleted file mode 100644
index 27a806907f5d..000000000000
--- a/buildtools/src/main/java/org/apache/pulsar/tests/MockitoCleanupListener.java
+++ /dev/null
@@ -1,64 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.pulsar.tests;
-
-import java.util.List;
-import lombok.CustomLog;
-import org.mockito.Mockito;
-import org.testng.ITestClass;
-
-/**
- * Cleanup Mockito's Thread Local state that leaks memory
- * Mockito.reset method should be called at the end of a test in the same thread where the methods were
- * mocked/stubbed. There are some tests which mock methods in the ForkJoinPool thread and these leak memory.
- * This listener doesn't support parallel execution at TestNG level. This is not thread safe.
- * Separate forks {@code testForkCount > 1} controlled with Maven Surefire is the recommended solution
- * for parallel test execution and that is fine.
- */
-@CustomLog
-public class MockitoCleanupListener extends BetweenTestClassesListenerAdapter {
- private static final boolean MOCKITO_CLEANUP_ENABLED = Boolean.parseBoolean(
- System.getProperty("testMockitoCleanup", "true"));
-
- private static final String MOCKITO_CLEANUP_INFO =
- "Cleaning up Mockito's ThreadSafeMockingProgress.MOCKING_PROGRESS_PROVIDER thread local state.";
-
- @Override
- protected void onBetweenTestClasses(List<ITestClass> testClasses) {
- if (MOCKITO_CLEANUP_ENABLED) {
- try {
- if (MockitoThreadLocalStateCleaner.INSTANCE.isEnabled()) {
- log.info(MOCKITO_CLEANUP_INFO);
- MockitoThreadLocalStateCleaner.INSTANCE.cleanup();
- }
- } finally {
- cleanupMockitoInline();
- }
- }
- }
-
- /**
- * Mockito-inline can leak mocked objects, we need to clean up the inline mocks after every test.
- * See <a href="https://javadoc.io/doc/org.mockito/mockito-core/latest/org/mockito/Mockito.html#47"}>
- * mockito docs</a>.
- */
- private void cleanupMockitoInline() {
- Mockito.framework().clearInlineMocks();
- }
-}
diff --git a/buildtools/src/test/java/org/apache/pulsar/tests/FastThreadLocalStateCleanerTest.java b/buildtools/src/test/java/org/apache/pulsar/tests/FastThreadLocalStateCleanerTest.java
deleted file mode 100644
index 93af1d647819..000000000000
--- a/buildtools/src/test/java/org/apache/pulsar/tests/FastThreadLocalStateCleanerTest.java
+++ /dev/null
@@ -1,128 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.pulsar.tests;
-
-import static org.testng.Assert.assertEquals;
-import static org.testng.Assert.assertTrue;
-import io.netty.util.concurrent.FastThreadLocal;
-import io.netty.util.concurrent.FastThreadLocalThread;
-import java.util.HashSet;
-import java.util.Set;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ExecutionException;
-import org.testng.annotations.Test;
-
-public class FastThreadLocalStateCleanerTest {
- private static final class MagicNumberWrapper {
- private final int value;
-
- private MagicNumberWrapper(int value) {
- this.value = value;
- }
-
- public int intValue() {
- return value;
- }
- }
-
- final FastThreadLocal<MagicNumberWrapper> magicNumberThreadLocal = new FastThreadLocal<MagicNumberWrapper>() {
- @Override
- protected MagicNumberWrapper initialValue() throws Exception {
- return new MagicNumberWrapper(42);
- }
- };
- final FastThreadLocalStateCleaner cleaner = new FastThreadLocalStateCleaner(object ->
- object.getClass() == MagicNumberWrapper.class);
-
- @Test
- public void testThreadLocalStateCleanupInCurrentThread() {
- magicNumberThreadLocal.set(new MagicNumberWrapper(44));
- assertEquals(magicNumberThreadLocal.get().intValue(), 44);
- cleaner.cleanupAllFastThreadLocals(Thread.currentThread(), ((thread, o) -> {
- System.out.println("Cleaning up " + thread + " value " + o);
- }));
- assertEquals(magicNumberThreadLocal.get().intValue(), 42);
- }
-
- @Test
- public void testThreadLocalStateCleanupInCurrentAndOtherThread() throws InterruptedException, ExecutionException {
- magicNumberThreadLocal.set(new MagicNumberWrapper(44));
- assertEquals(magicNumberThreadLocal.get().intValue(), 44);
-
- CountDownLatch numberHasBeenSet = new CountDownLatch(1);
- CountDownLatch shutdownLatch = new CountDownLatch(1);
- CompletableFuture<MagicNumberWrapper> valueAfterReset = new CompletableFuture<>();
- Thread thread = new Thread(() -> {
- try {
- magicNumberThreadLocal.set(new MagicNumberWrapper(45));
- assertEquals(magicNumberThreadLocal.get().intValue(), 45);
- numberHasBeenSet.countDown();
- shutdownLatch.await();
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new RuntimeException(e);
- } finally {
- valueAfterReset.complete(magicNumberThreadLocal.get());
- }
- });
- thread.start();
- numberHasBeenSet.await();
- Set<Thread> cleanedThreads = new HashSet<>();
- cleaner.cleanupAllFastThreadLocals((t, currentValue) -> {
- cleanedThreads.add(t);
- });
- shutdownLatch.countDown();
- assertEquals(magicNumberThreadLocal.get().intValue(), 42);
- assertEquals(valueAfterReset.get().intValue(), 42);
- assertEquals(cleanedThreads.size(), 2);
- assertTrue(cleanedThreads.contains(thread));
- assertTrue(cleanedThreads.contains(Thread.currentThread()));
- }
-
- @Test
- public void testThreadLocalStateCleanupInFastThreadLocalThread() throws InterruptedException, ExecutionException {
- CountDownLatch numberHasBeenSet = new CountDownLatch(1);
- CountDownLatch shutdownLatch = new CountDownLatch(1);
- CompletableFuture<MagicNumberWrapper> valueAfterReset = new CompletableFuture<>();
- Thread thread = new FastThreadLocalThread(() -> {
- try {
- magicNumberThreadLocal.set(new MagicNumberWrapper(45));
- assertEquals(magicNumberThreadLocal.get().intValue(), 45);
- numberHasBeenSet.countDown();
- shutdownLatch.await();
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new RuntimeException(e);
- } finally {
- valueAfterReset.complete(magicNumberThreadLocal.get());
- }
- });
- thread.start();
- numberHasBeenSet.await();
- Set<Thread> cleanedThreads = new HashSet<>();
- cleaner.cleanupAllFastThreadLocals((t, currentValue) -> {
- cleanedThreads.add(t);
- });
- shutdownLatch.countDown();
- assertEquals(valueAfterReset.get().intValue(), 42);
- assertTrue(cleanedThreads.contains(thread));
- }
-
-}
\ No newline at end of file
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
index bb1322b0b224..45f28847b94e 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
@@ -3236,18 +3236,8 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
}
}, null);
- CompletableFuture<Integer> batchSizeFuture = entryFuture.thenApply(entry -> {
- try {
- MessageMetadata metadata = entry.getMessageMetadata();
- if (metadata == null) {
- metadata = Commands.parseMessageMetadata(entry.getDataBuffer());
- }
- int batchSize = metadata.getNumMessagesInBatch();
- return metadata.hasNumMessagesInBatch() ? batchSize : -1;
- } finally {
- entry.release();
- }
- });
+ CompletableFuture<Integer> batchSizeFuture =
+ entryFuture.thenApply(ServerCnx::parseBatchSizeAndReleaseEntry);
batchSizeFuture.whenComplete((batchSize, e) -> {
if (e != null) {
@@ -3278,6 +3268,20 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
});
}
+ @VisibleForTesting
+ static int parseBatchSizeAndReleaseEntry(Entry entry) {
+ try {
+ MessageMetadata metadata = entry.getMessageMetadata();
+ if (metadata == null) {
+ metadata = Commands.parseMessageMetadata(entry.getDataBuffer());
+ }
+ int batchSize = metadata.getNumMessagesInBatch();
+ return metadata.hasNumMessagesInBatch() ? batchSize : -1;
+ } finally {
+ entry.release();
+ }
+ }
+
private void handleLastMessageIdFromCompactionService(PersistentTopic persistentTopic, long requestId,
int partitionIndex, Position markDeletePosition) {
persistentTopic.getTopicCompactionService().getLastMessagePosition().thenAccept(position ->
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
index 03742d9e7ff3..76127909423e 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -2407,7 +2407,8 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
}
}
- private void checkMessageExpiryWithoutSharedPosition(int messageTtlInSeconds) {
+ @VisibleForTesting
+ void checkMessageExpiryWithoutSharedPosition(int messageTtlInSeconds) {
subscriptions.forEach((__, sub) -> {
// TTL must not advance non-durable reader cursors past unread retained messages.
if (sub.getCursor().isDurable() && !isCompactionSubscription(sub.getName())
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/GetLastMessageIdEntryLeakTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/GetLastMessageIdEntryLeakTest.java
index 1c3bc3404049..c61ee906fe44 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/GetLastMessageIdEntryLeakTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/GetLastMessageIdEntryLeakTest.java
@@ -18,27 +18,18 @@
*/
package org.apache.pulsar.broker.service;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.doAnswer;
-import static org.mockito.Mockito.spy;
-import static org.testng.Assert.assertEquals;
-import static org.testng.Assert.assertThrows;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
-import lombok.Cleanup;
-import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntryCallback;
-import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.impl.EntryImpl;
-import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
-import org.apache.commons.lang3.reflect.FieldUtils;
-import org.apache.pulsar.broker.service.persistent.PersistentTopic;
-import org.apache.pulsar.client.api.Consumer;
-import org.apache.pulsar.client.api.Producer;
-import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.common.api.proto.MessageMetadata;
+import org.apache.pulsar.common.protocol.Commands;
+import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
@Test(groups = "broker")
-public class GetLastMessageIdEntryLeakTest extends SharedPulsarBaseTest {
+public class GetLastMessageIdEntryLeakTest {
/**
* Reproduces the ByteBuf leak in ServerCnx#getLargestBatchIndexWhenPossible: when
@@ -46,48 +37,51 @@ public class GetLastMessageIdEntryLeakTest extends SharedPulsarBaseTest {
* its backing ByteBuf) must still be released.
*/
@Test
- public void testEntryReleasedWhenParseMetadataThrows() throws Exception {
- final String topic = newTopicName();
-
- @Cleanup
- Producer<byte[]> producer = pulsarClient.newProducer().topic(topic).create();
- producer.send("payload".getBytes());
-
- @Cleanup
- Consumer<byte[]> consumer = pulsarClient.newConsumer(Schema.BYTES)
- .topic(topic).subscriptionName("sub").subscribe();
-
- PersistentTopic persistentTopic = (PersistentTopic) getTopicReference(topic).get();
-
+ public void testEntryReleasedWhenParseMetadataThrows() {
// A 2-byte, non-magic buffer: parseMessageMetadata reads it but fails at readUnsignedInt
// (needs 4 bytes), which is exactly the corrupt-entry case that triggers the leak.
- ByteBuf corruptBuf = Unpooled.buffer(2);
- corruptBuf.writeShort(0x0000);
-
- // Build the entry eagerly so its retain happens now; then drop our own ref so the entry
- // "owns" the only ref and we can observe whether the broker releases it.
- Position lastPosition = ((ManagedLedgerImpl) persistentTopic.getManagedLedger()).getLastConfirmedEntry();
- EntryImpl corruptEntry =
- EntryImpl.create(lastPosition.getLedgerId(), lastPosition.getEntryId(), corruptBuf);
- corruptBuf.release();
- assertEquals(corruptBuf.refCnt(), 1);
-
- // Spy the real ManagedLedgerImpl so asyncReadEntry hands back our corrupt entry, while
- // getLastPosition() etc. still delegate to the real ledger.
- ManagedLedgerImpl spyLedger = spy((ManagedLedgerImpl) persistentTopic.getManagedLedger());
- doAnswer(inv -> {
- ReadEntryCallback callback = inv.getArgument(1);
- callback.readEntryComplete(corruptEntry, inv.getArgument(2));
- return null;
- }).when(spyLedger).asyncReadEntry(any(Position.class), any(ReadEntryCallback.class), any());
-
- FieldUtils.writeField(persistentTopic, "ledger", spyLedger, true);
+ EntryImpl entry = EntryImpl.create(1, 0, new byte[2]);
+ ByteBuf buffer = entry.getDataBuffer();
+ try {
+ assertThatThrownBy(() -> ServerCnx.parseBatchSizeAndReleaseEntry(entry))
+ .isInstanceOf(IndexOutOfBoundsException.class);
+ assertThat(buffer.refCnt()).as("entry's ByteBuf after metadata parsing failed").isZero();
+ } finally {
+ if (buffer.refCnt() > 0) {
+ entry.release();
+ }
+ }
+ }
- // The broker fails the request (MetadataError) - that's expected; the point is the buffer.
- assertThrows(Exception.class, consumer::getLastMessageId);
+ @DataProvider
+ public Object[][] batchMetadata() {
+ return new Object[][] {{false, false}, {false, true}, {true, false}, {true, true}};
+ }
- // Before the fix: parseMessageMetadata throws, entry.release() is skipped -> refCnt stays 1.
- // After the fix: release runs in finally -> refCnt drops to 0.
- assertEquals(corruptBuf.refCnt(), 0, "entry's ByteBuf leaked when parseMessageMetadata threw");
+ @Test(dataProvider = "batchMetadata")
+ public void testEntryReleasedAfterReadingBatchSize(boolean cachedMetadata, boolean batched) {
+ MessageMetadata metadata = new MessageMetadata()
+ .setProducerName("producer")
+ .setSequenceId(0)
+ .setPublishTime(0);
+ if (batched) {
+ metadata.setNumMessagesInBatch(3);
+ }
+ // An invalid buffer with cached metadata also verifies that parsing the buffer is skipped.
+ ByteBuf buffer = cachedMetadata ? Unpooled.wrappedBuffer(new byte[2])
+ : Commands.serializeMetadataAndPayload(Commands.ChecksumType.None, metadata, Unpooled.EMPTY_BUFFER);
+ EntryImpl entry = EntryImpl.create(1, 0, buffer);
+ buffer.release();
+ if (cachedMetadata) {
+ entry.setMessageMetadata(metadata);
+ }
+ try {
+ assertThat(ServerCnx.parseBatchSizeAndReleaseEntry(entry)).isEqualTo(batched ? 3 : -1);
+ assertThat(buffer.refCnt()).as("entry's ByteBuf after reading batch size").isZero();
+ } finally {
+ if (buffer.refCnt() > 0) {
+ entry.release();
+ }
+ }
}
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersClassicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersClassicTest.java
index e81e4bb6c1cc..2c4f5729af89 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersClassicTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersClassicTest.java
@@ -30,18 +30,14 @@ import lombok.Cleanup;
import lombok.CustomLog;
import org.apache.bookkeeper.mledger.ManagedCursor;
import org.apache.bookkeeper.mledger.ManagedLedgerException;
-import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
import org.apache.pulsar.broker.service.Dispatcher;
import org.apache.pulsar.broker.service.SharedPulsarBaseTest;
-import org.apache.pulsar.broker.service.Subscription;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.MessageId;
-import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.common.api.proto.MessageMetadata;
import org.awaitility.reflect.WhiteboxImpl;
-import org.mockito.Mockito;
import org.testng.Assert;
import org.testng.annotations.Test;
@@ -56,6 +52,7 @@ public class PersistentDispatcherMultipleConsumersClassicTest extends SharedPuls
admin.topics().createNonPartitionedTopic(topicName);
admin.topics().createSubscription(topicName, subscription, MessageId.earliest);
+ @Cleanup
Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
.topic(topicName).subscriptionName(subscription)
.subscriptionType(SubscriptionType.Shared).subscribe();
@@ -83,6 +80,7 @@ public class PersistentDispatcherMultipleConsumersClassicTest extends SharedPuls
admin.topics().createNonPartitionedTopic(topicName);
admin.topics().createSubscription(topicName, subscription, MessageId.earliest);
+ @Cleanup
Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
.topic(topicName).subscriptionName(subscription)
.subscriptionType(SubscriptionType.Shared).subscribe();
@@ -100,62 +98,26 @@ public class PersistentDispatcherMultipleConsumersClassicTest extends SharedPuls
@Test
public void testSkipReadEntriesFromCloseCursor() throws Exception {
- final String topicName = newTopicName();
- final String subscription = "s1";
+ String topicName = newTopicName();
+ String subscription = "s1";
admin.topics().createNonPartitionedTopic(topicName);
-
- @Cleanup
- Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topicName).create();
- for (int i = 0; i < 10; i++) {
- producer.send("message-" + i);
- }
- producer.close();
-
- // Get the dispatcher of the topic.
- PersistentTopic topic = (PersistentTopic) getTopic(topicName, false).join().get();
-
- ManagedCursor cursor = Mockito.mock(ManagedCursorImpl.class);
- Mockito.doReturn(subscription).when(cursor).getName();
- Subscription sub = Mockito.mock(PersistentSubscription.class);
- Mockito.doReturn(topic).when(sub).getTopic();
- // Mock the dispatcher.
+ admin.topics().createSubscription(topicName, subscription, MessageId.earliest);
+ PersistentTopic topic = (PersistentTopic) getTopic(topicName, false).join().orElseThrow();
+ PersistentSubscription sub = topic.getSubscription(subscription);
+ AtomicInteger scheduledReads = new AtomicInteger();
PersistentDispatcherMultipleConsumersClassic dispatcher =
- Mockito.spy(new PersistentDispatcherMultipleConsumersClassic(topic, cursor, sub));
- // Return 10 permits to make the dispatcher can read more entries.
- Mockito.doReturn(10).when(dispatcher).getFirstAvailableConsumerPermits();
-
- // Make the count + 1 when call the scheduleReadEntriesWithDelay(...).
- AtomicInteger callScheduleReadEntriesWithDelayCnt = new AtomicInteger(0);
- Mockito.doAnswer(inv -> {
- callScheduleReadEntriesWithDelayCnt.getAndIncrement();
- return inv.callRealMethod();
- }).when(dispatcher).scheduleReadEntriesWithDelay(Mockito.any(), Mockito.any(), Mockito.anyLong());
-
- // Make the count + 1 when call the readEntriesFailed(...).
- AtomicInteger callReadEntriesFailed = new AtomicInteger(0);
- Mockito.doAnswer(inv -> {
- callReadEntriesFailed.getAndIncrement();
- return inv.callRealMethod();
- }).when(dispatcher).readEntriesFailed(Mockito.any(), Mockito.any());
-
- Mockito.doReturn(false).when(cursor).isClosed();
-
- // Mock the readEntriesOrWait(...) to simulate the cursor is closed.
- Mockito.doAnswer(inv -> {
- PersistentDispatcherMultipleConsumersClassic dispatcher1 = inv.getArgument(2);
- dispatcher1.readEntriesFailed(new ManagedLedgerException.CursorAlreadyClosedException("cursor closed"),
- null);
- return null;
- }).when(cursor).asyncReadEntriesOrWait(Mockito.anyInt(), Mockito.anyLong(), Mockito.eq(dispatcher),
- Mockito.any(), Mockito.any());
-
- dispatcher.readMoreEntries();
-
- // Verify: the readEntriesFailed should be called once and
- // the scheduleReadEntriesWithDelay should not be called.
- Assert.assertTrue(callReadEntriesFailed.get() == 1 && callScheduleReadEntriesWithDelayCnt.get() == 0);
+ new PersistentDispatcherMultipleConsumersClassic(topic, sub.getCursor(), sub) {
+ @Override
+ void scheduleReadEntriesWithDelay(Exception exception, ReadType readType, long delay) {
+ scheduledReads.incrementAndGet();
+ super.scheduleReadEntriesWithDelay(exception, readType, delay);
+ }
+ };
- // Verify: the topic can be deleted successfully.
+ dispatcher.readEntriesFailed(new ManagedLedgerException.CursorAlreadyClosedException("cursor closed"),
+ null);
+
+ Assert.assertEquals(scheduledReads.get(), 0, "Closed cursor failures must not schedule another read");
admin.topics().delete(topicName, false);
}
@@ -172,17 +134,15 @@ public class PersistentDispatcherMultipleConsumersClassicTest extends SharedPuls
final String subscription = "s1";
// Needed to create the topic
+ @Cleanup
Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
.topic(topicName).subscriptionName(subscription)
.subscriptionType(SubscriptionType.Shared).subscribe();
PersistentTopic topic = (PersistentTopic) getTopic(topicName, false).join().get();
- ManagedCursor cursor = Mockito.mock(ManagedCursorImpl.class);
- Mockito.doReturn(subscription).when(cursor).getName();
-
- Subscription sub = Mockito.mock(PersistentSubscription.class);
- Mockito.doReturn(topic).when(sub).getTopic();
+ PersistentSubscription sub = topic.getSubscription(subscription);
+ ManagedCursor cursor = sub.getCursor();
PersistentDispatcherMultipleConsumersClassic dispatcher =
new PersistentDispatcherMultipleConsumersClassic(topic, cursor, sub);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersTest.java
index daea92e0dcd8..2c6d97d58013 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersTest.java
@@ -21,6 +21,7 @@ package org.apache.pulsar.broker.service.persistent;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import com.carrotsearch.hppc.ObjectSet;
+import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
@@ -33,29 +34,46 @@ import lombok.Cleanup;
import lombok.CustomLog;
import org.apache.bookkeeper.mledger.ManagedCursor;
import org.apache.bookkeeper.mledger.ManagedLedgerException;
-import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
import org.apache.pulsar.broker.service.Dispatcher;
import org.apache.pulsar.broker.service.SharedPulsarBaseTest;
import org.apache.pulsar.broker.service.Subscription;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.MessageId;
-import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.common.api.proto.MessageMetadata;
import org.awaitility.reflect.WhiteboxImpl;
import org.mockito.Mockito;
import org.testng.Assert;
+import org.testng.annotations.AfterMethod;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
@CustomLog
@Test(groups = "broker-api")
public class PersistentDispatcherMultipleConsumersTest extends SharedPulsarBaseTest {
+ // Concurrent mock calls only run on executors that the test shuts down and joins.
+ private final List<Object> ownedMocks = new ArrayList<>();
+
+ private <T> T ownMock(Class<T> type) {
+ T mock = Mockito.mock(type);
+ ownedMocks.add(mock);
+ return mock;
+ }
+
+ @AfterMethod(alwaysRun = true)
+ public void releaseOwnedMocks() {
+ try {
+ Mockito.reset(ownedMocks.toArray());
+ } finally {
+ ownedMocks.forEach(mock -> Mockito.framework().clearInlineMock(mock));
+ ownedMocks.clear();
+ }
+ }
@Test(timeOut = 30_000, dataProvider = "readConflationDispatcherTypes")
public void testReadMoreEntriesConflatesConcurrentRequests(boolean classic) throws Exception {
- ManagedCursor cursor = Mockito.mock(ManagedCursor.class);
+ ManagedCursor cursor = ownMock(ManagedCursor.class);
AbstractPersistentDispatcherMultipleConsumers dispatcher = createReadConflationDispatcher(cursor, classic);
CountDownLatch[] passStarted = {new CountDownLatch(1), new CountDownLatch(1)};
CountDownLatch[] releasePass = {new CountDownLatch(1), new CountDownLatch(1)};
@@ -99,7 +117,7 @@ public class PersistentDispatcherMultipleConsumersTest extends SharedPulsarBaseT
@Test(timeOut = 30_000, dataProvider = "readConflationDispatcherTypes")
public void testReadMoreEntriesConflatesReentrantRequests(boolean classic) throws Exception {
- ManagedCursor cursor = Mockito.mock(ManagedCursor.class);
+ ManagedCursor cursor = ownMock(ManagedCursor.class);
AbstractPersistentDispatcherMultipleConsumers dispatcher = createReadConflationDispatcher(cursor, classic);
AtomicInteger passes = new AtomicInteger();
AtomicInteger depth = new AtomicInteger();
@@ -122,7 +140,7 @@ public class PersistentDispatcherMultipleConsumersTest extends SharedPulsarBaseT
@Test(timeOut = 30_000, dataProvider = "readConflationDispatcherTypes")
public void testReadMoreEntriesRecoversAfterFailure(boolean classic) throws Exception {
- ManagedCursor cursor = Mockito.mock(ManagedCursor.class);
+ ManagedCursor cursor = ownMock(ManagedCursor.class);
AbstractPersistentDispatcherMultipleConsumers dispatcher = createReadConflationDispatcher(cursor, classic);
IllegalStateException failure = new IllegalStateException("read failed");
Mockito.doThrow(failure).when(cursor).isClosed();
@@ -136,7 +154,7 @@ public class PersistentDispatcherMultipleConsumersTest extends SharedPulsarBaseT
@Test(timeOut = 30_000, dataProvider = "readConflationDispatcherTypes")
public void testReadMoreEntriesRunsReentrantRequestsAfterFailure(boolean classic) throws Exception {
- ManagedCursor cursor = Mockito.mock(ManagedCursor.class);
+ ManagedCursor cursor = ownMock(ManagedCursor.class);
AbstractPersistentDispatcherMultipleConsumers dispatcher = createReadConflationDispatcher(cursor, classic);
IllegalStateException failure = new IllegalStateException("read failed");
IllegalStateException followUpFailure = new IllegalStateException("follow-up read failed");
@@ -160,7 +178,7 @@ public class PersistentDispatcherMultipleConsumersTest extends SharedPulsarBaseT
@Test(timeOut = 30_000, dataProvider = "readConflationDispatcherTypes")
public void testReadMoreEntriesRunsConcurrentRequestAfterFailure(boolean classic) throws Exception {
- ManagedCursor cursor = Mockito.mock(ManagedCursor.class);
+ ManagedCursor cursor = ownMock(ManagedCursor.class);
AbstractPersistentDispatcherMultipleConsumers dispatcher = createReadConflationDispatcher(cursor, classic);
IllegalStateException failure = new IllegalStateException("read failed");
CountDownLatch passStarted = new CountDownLatch(1);
@@ -205,7 +223,7 @@ public class PersistentDispatcherMultipleConsumersTest extends SharedPulsarBaseT
admin.topics().createSubscription(topicName, "s1", MessageId.earliest);
PersistentTopic topic = (PersistentTopic) getTopic(topicName, false).join().orElseThrow();
Mockito.doReturn("s1").when(cursor).getName();
- Subscription subscription = Mockito.mock(PersistentSubscription.class);
+ Subscription subscription = ownMock(PersistentSubscription.class);
Mockito.doReturn(topic).when(subscription).getTopic();
return classic ? new PersistentDispatcherMultipleConsumersClassic(topic, cursor, subscription)
: new PersistentDispatcherMultipleConsumers(topic, cursor, subscription);
@@ -218,6 +236,7 @@ public class PersistentDispatcherMultipleConsumersTest extends SharedPulsarBaseT
admin.topics().createNonPartitionedTopic(topicName);
admin.topics().createSubscription(topicName, subscription, MessageId.earliest);
+ @Cleanup
Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
.topic(topicName).subscriptionName(subscription)
.subscriptionType(SubscriptionType.Shared).subscribe();
@@ -245,6 +264,7 @@ public class PersistentDispatcherMultipleConsumersTest extends SharedPulsarBaseT
admin.topics().createNonPartitionedTopic(topicName);
admin.topics().createSubscription(topicName, subscription, MessageId.earliest);
+ @Cleanup
Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
.topic(topicName).subscriptionName(subscription)
.subscriptionType(SubscriptionType.Shared).subscribe();
@@ -262,62 +282,26 @@ public class PersistentDispatcherMultipleConsumersTest extends SharedPulsarBaseT
@Test
public void testSkipReadEntriesFromCloseCursor() throws Exception {
- final String topicName = newTopicName();
- final String subscription = "s1";
+ String topicName = newTopicName();
+ String subscription = "s1";
admin.topics().createNonPartitionedTopic(topicName);
-
- @Cleanup
- Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topicName).create();
- for (int i = 0; i < 10; i++) {
- producer.send("message-" + i);
- }
- producer.close();
-
- // Get the dispatcher of the topic.
- PersistentTopic topic = (PersistentTopic) getTopic(topicName, false).join().get();
-
- ManagedCursor cursor = Mockito.mock(ManagedCursorImpl.class);
- Mockito.doReturn(subscription).when(cursor).getName();
- Subscription sub = Mockito.mock(PersistentSubscription.class);
- Mockito.doReturn(topic).when(sub).getTopic();
- // Mock the dispatcher.
+ admin.topics().createSubscription(topicName, subscription, MessageId.earliest);
+ PersistentTopic topic = (PersistentTopic) getTopic(topicName, false).join().orElseThrow();
+ PersistentSubscription sub = topic.getSubscription(subscription);
+ AtomicInteger scheduledReads = new AtomicInteger();
PersistentDispatcherMultipleConsumers dispatcher =
- Mockito.spy(new PersistentDispatcherMultipleConsumers(topic, cursor, sub));
- // Return 10 permits to make the dispatcher can read more entries.
- Mockito.doReturn(10).when(dispatcher).getFirstAvailableConsumerPermits();
-
- // Make the count + 1 when call the scheduleReadEntriesWithDelay(...).
- AtomicInteger callScheduleReadEntriesWithDelayCnt = new AtomicInteger(0);
- Mockito.doAnswer(inv -> {
- callScheduleReadEntriesWithDelayCnt.getAndIncrement();
- return inv.callRealMethod();
- }).when(dispatcher).scheduleReadEntriesWithDelay(Mockito.any(), Mockito.any(), Mockito.anyLong());
-
- // Make the count + 1 when call the readEntriesFailed(...).
- AtomicInteger callReadEntriesFailed = new AtomicInteger(0);
- Mockito.doAnswer(inv -> {
- callReadEntriesFailed.getAndIncrement();
- return inv.callRealMethod();
- }).when(dispatcher).readEntriesFailed(Mockito.any(), Mockito.any());
-
- Mockito.doReturn(false).when(cursor).isClosed();
-
- // Mock the readEntriesOrWait(...) to simulate the cursor is closed.
- Mockito.doAnswer(inv -> {
- AbstractPersistentDispatcherMultipleConsumers dispatcher1 = inv.getArgument(2);
- dispatcher1.readEntriesFailed(new ManagedLedgerException.CursorAlreadyClosedException("cursor closed"),
- null);
- return null;
- }).when(cursor).asyncReadEntriesWithSkipOrWait(Mockito.anyInt(), Mockito.anyLong(), Mockito.eq(dispatcher),
- Mockito.any(), Mockito.any(), Mockito.any());
-
- dispatcher.readMoreEntries();
+ new PersistentDispatcherMultipleConsumers(topic, sub.getCursor(), sub) {
+ @Override
+ void scheduleReadEntriesWithDelay(Exception exception, ReadType readType, long delay) {
+ scheduledReads.incrementAndGet();
+ super.scheduleReadEntriesWithDelay(exception, readType, delay);
+ }
+ };
- // Verify: the readEntriesFailed should be called once and
- // the scheduleReadEntriesWithDelay should not be called.
- Assert.assertTrue(callReadEntriesFailed.get() == 1 && callScheduleReadEntriesWithDelayCnt.get() == 0);
+ dispatcher.readEntriesFailed(new ManagedLedgerException.CursorAlreadyClosedException("cursor closed"),
+ null);
- // Verify: the topic can be deleted successfully.
+ Assert.assertEquals(scheduledReads.get(), 0, "Closed cursor failures must not schedule another read");
admin.topics().delete(topicName, false);
}
@@ -334,17 +318,15 @@ public class PersistentDispatcherMultipleConsumersTest extends SharedPulsarBaseT
final String subscription = "s1";
// Needed to create the topic
+ @Cleanup
Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
.topic(topicName).subscriptionName(subscription)
.subscriptionType(SubscriptionType.Shared).subscribe();
PersistentTopic topic = (PersistentTopic) getTopic(topicName, false).join().get();
- ManagedCursor cursor = Mockito.mock(ManagedCursorImpl.class);
- Mockito.doReturn(subscription).when(cursor).getName();
-
- Subscription sub = Mockito.mock(PersistentSubscription.class);
- Mockito.doReturn(topic).when(sub).getTopic();
+ PersistentSubscription sub = topic.getSubscription(subscription);
+ ManagedCursor cursor = sub.getCursor();
PersistentDispatcherMultipleConsumers dispatcher =
new PersistentDispatcherMultipleConsumers(topic, cursor, sub);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java
index 232f127d4747..de95404425c6 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java
@@ -18,7 +18,9 @@
*/
package org.apache.pulsar.broker.service.persistent;
+import java.util.Collections;
import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
@@ -27,24 +29,19 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import lombok.Cleanup;
import lombok.CustomLog;
-import org.apache.bookkeeper.mledger.AsyncCallbacks;
-import org.apache.bookkeeper.mledger.ManagedCursor;
import org.apache.bookkeeper.mledger.ManagedLedgerException;
-import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
+import org.apache.pulsar.broker.intercept.BrokerInterceptor;
import org.apache.pulsar.broker.intercept.MockBrokerInterceptor;
+import org.apache.pulsar.broker.service.BrokerService;
import org.apache.pulsar.broker.service.BrokerServiceException;
import org.apache.pulsar.broker.service.Consumer;
import org.apache.pulsar.broker.service.ServerCnx;
import org.apache.pulsar.broker.service.SharedPulsarBaseTest;
import org.apache.pulsar.broker.service.SharedPulsarCluster;
-import org.apache.pulsar.broker.service.Subscription;
-import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.PulsarClient;
-import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.common.api.proto.CommandSubscribe;
import org.apache.pulsar.common.naming.TopicName;
-import org.awaitility.Awaitility;
-import org.mockito.Mockito;
import org.testng.Assert;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
@@ -55,74 +52,27 @@ public class PersistentDispatcherSingleActiveConsumerTest extends SharedPulsarBa
@Test
public void testSkipReadEntriesFromCloseCursor() throws Exception {
- final String topicName = newTopicName();
- final String subscription = "s1";
+ String topicName = newTopicName();
+ String subscription = "s1";
admin.topics().createNonPartitionedTopic(topicName);
-
- @Cleanup
- Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topicName).create();
- for (int i = 0; i < 10; i++) {
- producer.send("message-" + i);
- }
- producer.close();
-
- // Get the dispatcher of the topic.
- PersistentTopic topic = (PersistentTopic) getTopic(topicName, false).join().get();
-
- ManagedCursor cursor = Mockito.mock(ManagedCursorImpl.class);
- Mockito.doReturn(subscription).when(cursor).getName();
- Subscription sub = Mockito.mock(PersistentSubscription.class);
- Mockito.doReturn(topic).when(sub).getTopic();
- // Mock the dispatcher.
+ admin.topics().createSubscription(topicName, subscription, MessageId.earliest);
+ PersistentTopic topic = (PersistentTopic) getTopic(topicName, false).join().orElseThrow();
+ PersistentSubscription sub = topic.getSubscription(subscription);
+ AtomicInteger scheduledReads = new AtomicInteger();
PersistentDispatcherSingleActiveConsumer dispatcher =
- Mockito.spy(new PersistentDispatcherSingleActiveConsumer(cursor,
- CommandSubscribe.SubType.Exclusive, 0, topic, sub));
-
- // Mock a consumer
- Consumer consumer = Mockito.mock(Consumer.class);
- consumer.getAvailablePermits();
- Mockito.doReturn(10).when(consumer).getAvailablePermits();
- Mockito.doReturn(10).when(consumer).getAvgMessagesPerEntry();
- Mockito.doReturn("test").when(consumer).consumerName();
- Mockito.doReturn(true).when(consumer).isWritable();
- Mockito.doReturn(false).when(consumer).readCompacted();
-
- // Make the consumer as the active consumer.
- Mockito.doReturn(consumer).when(dispatcher).getActiveConsumer();
-
- // Make the count + 1 when call the scheduleReadEntriesWithDelay(...).
- AtomicInteger callScheduleReadEntriesWithDelayCnt = new AtomicInteger(0);
- Mockito.doAnswer(inv -> {
- callScheduleReadEntriesWithDelayCnt.getAndIncrement();
- return inv.callRealMethod();
- }).when(dispatcher).scheduleReadEntriesWithDelay(Mockito.eq(consumer), Mockito.anyLong());
-
- // Make the count + 1 when call the readEntriesFailed(...).
- AtomicInteger callReadEntriesFailed = new AtomicInteger(0);
- Mockito.doAnswer(inv -> {
- callReadEntriesFailed.getAndIncrement();
- return inv.callRealMethod();
- }).when(dispatcher).readEntriesFailed(Mockito.any(), Mockito.any(), Mockito.anyLong());
-
- Mockito.doReturn(false).when(cursor).isClosed();
-
- // Mock the readEntriesOrWait(...) to simulate the cursor is closed.
- Mockito.doAnswer(inv -> {
- final var callback = (AsyncCallbacks.ReadEntriesCallback) inv.getArgument(2);
- callback.readEntriesFailed(new ManagedLedgerException.CursorAlreadyClosedException("cursor closed"),
- null);
- return null;
- }).when(cursor).asyncReadEntriesWithSkipOrWait(Mockito.anyInt(), Mockito.anyLong(), Mockito.any(),
- Mockito.any(), Mockito.any(), Mockito.any());
-
- dispatcher.readMoreEntries(consumer);
-
- // Verify: the readEntriesFailed should be called once and
- // the scheduleReadEntriesWithDelay should not be called.
- Awaitility.await().untilAsserted(() -> Assert.assertTrue(callReadEntriesFailed.get() == 1
- && callScheduleReadEntriesWithDelayCnt.get() == 0));
-
- // Verify: the topic can be deleted successfully.
+ new PersistentDispatcherSingleActiveConsumer(sub.getCursor(),
+ CommandSubscribe.SubType.Exclusive, 0, topic, sub) {
+ @Override
+ void scheduleReadEntriesWithDelay(Consumer consumer, long delay) {
+ scheduledReads.incrementAndGet();
+ super.scheduleReadEntriesWithDelay(consumer, delay);
+ }
+ };
+
+ dispatcher.readEntriesFailed(new ManagedLedgerException.CursorAlreadyClosedException("cursor closed"),
+ null, 0);
+
+ Assert.assertEquals(scheduledReads.get(), 0, "Closed cursor failures must not schedule another read");
admin.topics().delete(topicName, false);
}
@@ -133,46 +83,64 @@ public class PersistentDispatcherSingleActiveConsumerTest extends SharedPulsarBa
@Test(dataProvider = "closeDelayMs")
public void testOverrideInactiveConsumer(long closeDelayMs) throws Exception {
+ BrokerService broker = SharedPulsarCluster.get().getPulsarService().getBrokerService();
+ BrokerInterceptor previousInterceptor = broker.getInterceptor();
final var interceptor = new Interceptor();
- SharedPulsarCluster.get().getPulsarService().getBrokerService().setInterceptor(interceptor);
- final var topic = newTopicName();
- @Cleanup final var client = PulsarClient.builder()
- .serviceUrl(getBrokerServiceUrl()).build();
- @Cleanup final var consumer = client.newConsumer().topic(topic).subscriptionName("sub").subscribe();
- final var dispatcher = ((PersistentTopic) SharedPulsarCluster.get().getPulsarService().getBrokerService()
- .getTopicIfExists(TopicName.get(topic).toString()).get().orElseThrow())
- .getSubscription("sub").dispatcher;
- Assert.assertEquals(dispatcher.getConsumers().size(), 1);
+ broker.setInterceptor(interceptor);
+ try {
+ final var topic = newTopicName();
+ @Cleanup final var client = PulsarClient.builder()
+ .serviceUrl(getBrokerServiceUrl()).build();
+ @Cleanup final var consumer = client.newConsumer().topic(topic).subscriptionName("sub").subscribe();
+ final var dispatcher = ((PersistentTopic) SharedPulsarCluster.get().getPulsarService().getBrokerService()
+ .getTopicIfExists(TopicName.get(topic).toString()).get().orElseThrow())
+ .getSubscription("sub").dispatcher;
+ Assert.assertEquals(dispatcher.getConsumers().size(), 1);
- // Generally `isActive` could only be false after `channelInactive` is called, setting it with false directly
- // to avoid race condition.
- final var latch = new CountDownLatch(1);
- interceptor.latch.set(latch);
- interceptor.injectCloseLatency.set(true);
- interceptor.delayMs = closeDelayMs;
- // Simulate the real case because `channelInactive` is always called in the event loop thread
- final var cnx = (ServerCnx) dispatcher.getConsumers().get(0).cnx();
- cnx.ctx().executor().execute(() -> {
- try {
- cnx.channelInactive(cnx.ctx());
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- });
+ // Generally `isActive` could only be false after `channelInactive` is called; set it with false directly
+ // to avoid race condition.
+ final var latch = new CountDownLatch(1);
+ interceptor.latch.set(latch);
+ interceptor.injectCloseLatency.set(true);
+ interceptor.delayMs = closeDelayMs;
+ // Simulate the real case because `channelInactive` is always called in the event loop thread
+ Consumer original = dispatcher.getConsumers().get(0);
+ final var cnx = (ServerCnx) original.cnx();
+ CompletableFuture<Void> connectionClosed = new CompletableFuture<>();
+ cnx.ctx().executor().execute(() -> {
+ try {
+ cnx.channelInactive(cnx.ctx());
+ connectionClosed.complete(null);
+ } catch (Exception e) {
+ connectionClosed.completeExceptionally(e);
+ }
+ });
- @Cleanup final var mockConsumer = Mockito.mock(Consumer.class);
- Assert.assertTrue(latch.await(1, TimeUnit.SECONDS));
- if (closeDelayMs < 1000) {
- dispatcher.addConsumer(mockConsumer).get();
- Assert.assertEquals(dispatcher.getConsumers().size(), 1);
- Assert.assertSame(mockConsumer, dispatcher.getConsumers().get(0));
- } else {
+ Consumer replacement = new Consumer(original.getSubscription(), original.subType(), topic,
+ original.consumerId() + 1, 0, "replacement", true, cnx, "role", Collections.emptyMap(), false,
+ null, MessageId.latest, 0);
try {
- dispatcher.addConsumer(mockConsumer).get();
- Assert.fail();
- } catch (ExecutionException e) {
- Assert.assertTrue(e.getCause() instanceof BrokerServiceException.ConsumerBusyException);
+ Assert.assertTrue(latch.await(1, TimeUnit.SECONDS));
+ if (closeDelayMs < 1000) {
+ dispatcher.addConsumer(replacement).get();
+ Assert.assertEquals(dispatcher.getConsumers().size(), 1);
+ Assert.assertSame(replacement, dispatcher.getConsumers().get(0));
+ } else {
+ try {
+ dispatcher.addConsumer(replacement).get();
+ Assert.fail();
+ } catch (ExecutionException e) {
+ Assert.assertTrue(e.getCause() instanceof BrokerServiceException.ConsumerBusyException);
+ }
+ }
+ } finally {
+ connectionClosed.get(10, TimeUnit.SECONDS);
+ if (dispatcher.getConsumers().contains(replacement)) {
+ dispatcher.removeConsumer(replacement);
+ }
}
+ } finally {
+ broker.setInterceptor(previousInterceptor);
}
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ReaderMessageTTLTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ReaderMessageTTLTest.java
index 2e57a883aca0..b4511494d5aa 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ReaderMessageTTLTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ReaderMessageTTLTest.java
@@ -19,12 +19,7 @@
package org.apache.pulsar.broker.service.persistent;
import static org.assertj.core.api.Assertions.assertThat;
-import static org.mockito.AdditionalAnswers.delegatesTo;
-import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.spy;
import java.util.concurrent.TimeUnit;
-import org.apache.bookkeeper.mledger.ManagedLedger;
import org.apache.bookkeeper.mledger.Position;
import org.apache.pulsar.broker.service.SharedPulsarBaseTest;
import org.apache.pulsar.client.api.Message;
@@ -73,16 +68,12 @@ public class ReaderMessageTTLTest extends SharedPulsarBaseTest {
// Pause the reader until all published messages are old enough to expire.
long expiryTime = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(2);
Awaitility.await().until(() -> System.currentTimeMillis() > expiryTime);
- PersistentTopic expiryTopic = topic;
- if (!sharedPosition) {
- expiryTopic = spy(topic);
- // Exercise the fallback used by custom ManagedLedger implementations.
- doReturn(mock(ManagedLedger.class, delegatesTo(topic.getManagedLedger())))
- .when(expiryTopic).getManagedLedger();
- }
- PersistentTopic topicToCheck = expiryTopic;
Awaitility.await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
- topicToCheck.checkMessageExpiry();
+ if (sharedPosition) {
+ topic.checkMessageExpiry();
+ } else {
+ topic.checkMessageExpiryWithoutSharedPosition(1);
+ }
assertThat(durable.getCursor().getNumberOfEntriesInBacklog(false)).isZero();
});
assertThat(readerSubscription.getCursor().getMarkDeletedPosition())
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/SharedDispatcherPermitAccountingTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/SharedDispatcherPermitAccountingTest.java
index 36f7bf900112..2ca880f5197e 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/SharedDispatcherPermitAccountingTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/SharedDispatcherPermitAccountingTest.java
@@ -28,26 +28,27 @@ import static org.apache.pulsar.common.protocol.Commands.DEFAULT_CONSUMER_EPOCH;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
-import static org.mockito.ArgumentMatchers.anyList;
-import static org.mockito.ArgumentMatchers.anyLong;
-import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.doThrow;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+import io.netty.channel.ChannelPromise;
+import io.netty.channel.DefaultChannelPromise;
import io.netty.channel.EventLoopGroup;
+import io.netty.channel.embedded.EmbeddedChannel;
import io.netty.util.concurrent.EventExecutor;
import io.netty.util.concurrent.ImmediateEventExecutor;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
+import java.util.Objects;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
import org.apache.bookkeeper.mledger.Entry;
import org.apache.bookkeeper.mledger.ManagedCursor;
+import org.apache.bookkeeper.mledger.PositionFactory;
+import org.apache.bookkeeper.mledger.impl.EntryImpl;
import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
import org.apache.pulsar.broker.service.BrokerService;
import org.apache.pulsar.broker.service.Consumer;
@@ -55,13 +56,14 @@ import org.apache.pulsar.broker.service.Dispatcher;
import org.apache.pulsar.broker.service.EntryBatchIndexesAcks;
import org.apache.pulsar.broker.service.EntryBatchSizes;
import org.apache.pulsar.broker.service.PulsarCommandSender;
+import org.apache.pulsar.broker.service.PulsarCommandSenderImpl;
import org.apache.pulsar.broker.service.RedeliveryTracker;
+import org.apache.pulsar.broker.service.RedeliveryTrackerDisabled;
import org.apache.pulsar.broker.service.ServerCnx;
import org.apache.pulsar.broker.service.SharedPulsarBaseTest;
import org.apache.pulsar.broker.service.StickyKeyDispatcher;
import org.apache.pulsar.broker.service.Subscription;
import org.apache.pulsar.broker.service.Topic;
-import org.apache.pulsar.broker.service.TransportCnx;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.Producer;
@@ -71,11 +73,55 @@ import org.apache.pulsar.common.api.proto.CommandSubscribe.SubType;
import org.apache.pulsar.common.api.proto.KeySharedMeta;
import org.apache.pulsar.common.policies.data.stats.ConsumerStatsImpl;
import org.awaitility.Awaitility;
+import org.mockito.Mockito;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
@Test(groups = "broker-api")
public class SharedDispatcherPermitAccountingTest extends SharedPulsarBaseTest {
+ private EmbeddedChannel channel;
+
+ @BeforeMethod(alwaysRun = true)
+ public void createPromiseChannel() {
+ channel = new EmbeddedChannel();
+ }
+
+ @AfterMethod(alwaysRun = true)
+ public void closePromiseChannel() {
+ if (channel != null) {
+ channel.finishAndReleaseAll();
+ channel = null;
+ }
+ }
+
+ /** Owns mocks confined to the calling test thread; shared workers use the real fixtures below. */
+ private static class MockScope implements AutoCloseable {
+ private final List<Object> mocks = new ArrayList<>();
+
+ <T> T mock(Class<T> type) {
+ T mock = Mockito.mock(type);
+ mocks.add(mock);
+ return mock;
+ }
+
+ <T> T spy(T object) {
+ T spy = Mockito.spy(object);
+ mocks.add(spy);
+ return spy;
+ }
+
+ @Override
+ public void close() {
+ try {
+ Mockito.reset(mocks.toArray());
+ } finally {
+ mocks.forEach(mock -> Mockito.framework().clearInlineMock(mock));
+ mocks.clear();
+ }
+ }
+ }
@DataProvider(name = "dispatcherImplementations")
public Object[][] dispatcherImplementations() {
@@ -105,25 +151,27 @@ public class SharedDispatcherPermitAccountingTest extends SharedPulsarBaseTest {
@Test(dataProvider = "untrackedSubscriptionVariants", timeOut = 30_000)
public void testUnrelatedSubscriptionDoesNotAccumulatePendingDispatcherPermits(
boolean persistent, SubType subType) throws Exception {
- String topicName = newTopicName();
- String subscriptionName = "sub";
- admin.topics().createNonPartitionedTopic(topicName);
- PersistentTopic persistentTopic = (PersistentTopic) getTopic(topicName, false).join().orElseThrow();
- Topic consumerTopic = persistent ? persistentTopic : mock(Topic.class);
- if (!persistent) {
- when(consumerTopic.getBrokerService()).thenReturn(persistentTopic.getBrokerService());
- when(consumerTopic.getHierarchyTopicPolicies())
- .thenReturn(persistentTopic.getHierarchyTopicPolicies());
- }
- Subscription subscription = mock(Subscription.class);
- when(subscription.getName()).thenReturn(subscriptionName);
- when(subscription.getTopic()).thenReturn(consumerTopic);
- Consumer consumer = createConsumer(subscription, subType, topicName, 1);
+ try (MockScope mocks = new MockScope()) {
+ String topicName = newTopicName();
+ String subscriptionName = "sub";
+ admin.topics().createNonPartitionedTopic(topicName);
+ PersistentTopic persistentTopic = (PersistentTopic) getTopic(topicName, false).join().orElseThrow();
+ Topic consumerTopic = persistent ? persistentTopic : mocks.mock(Topic.class);
+ if (!persistent) {
+ when(consumerTopic.getBrokerService()).thenReturn(persistentTopic.getBrokerService());
+ when(consumerTopic.getHierarchyTopicPolicies())
+ .thenReturn(persistentTopic.getHierarchyTopicPolicies());
+ }
+ Subscription subscription = mocks.mock(Subscription.class);
+ when(subscription.getName()).thenReturn(subscriptionName);
+ when(subscription.getTopic()).thenReturn(consumerTopic);
+ Consumer consumer = createConsumer(subscription, subType, topicName, 1);
- consumer.flowPermits(100);
+ consumer.flowPermits(100);
- assertThat(consumer.getAvailablePermits()).isEqualTo(100);
- assertThat(consumer.getAvailablePermitsForDispatcherRemoval()).isEqualTo(100);
+ assertThat(consumer.getAvailablePermits()).isEqualTo(100);
+ assertThat(consumer.getAvailablePermitsForDispatcherRemoval()).isEqualTo(100);
+ }
}
@Test(dataProvider = "flowRaceDispatcherVariants", timeOut = 30_000)
@@ -183,75 +231,79 @@ public class SharedDispatcherPermitAccountingTest extends SharedPulsarBaseTest {
@Test(dataProvider = "dispatcherImplementations", timeOut = 30_000)
public void testRejectedFlowStaysPendingAndIsExcludedFromRemoval(boolean classic) throws Exception {
- String topicName = newTopicName();
- String subscriptionName = "shared-sub";
- admin.topics().createNonPartitionedTopic(topicName);
- PersistentTopic realTopic = (PersistentTopic) getTopic(topicName, false).join().orElseThrow();
- BrokerService isolatedBroker = spy(realTopic.getBrokerService());
- PersistentTopic isolatedTopic = spy(realTopic);
- doReturn(isolatedBroker).when(isolatedTopic).getBrokerService();
-
- EventLoopGroup rejectingExecutor = mock(EventLoopGroup.class);
- doThrow(new RejectedExecutionException("test rejection"))
- .when(rejectingExecutor).execute(any(Runnable.class));
- when(rejectingExecutor.isShuttingDown()).thenReturn(true);
- doReturn(rejectingExecutor).when(isolatedBroker).executor();
-
- ManagedCursor cursor = mock(ManagedCursorImpl.class);
- when(cursor.getName()).thenReturn(subscriptionName);
- when(cursor.isClosed()).thenReturn(true);
- Subscription subscription = mock(PersistentSubscription.class);
- when(subscription.getName()).thenReturn(subscriptionName);
- when(subscription.getTopic()).thenReturn(isolatedTopic);
- Dispatcher dispatcher = classic
- ? new PersistentDispatcherMultipleConsumersClassic(isolatedTopic, cursor, subscription)
- : new PersistentDispatcherMultipleConsumers(isolatedTopic, cursor, subscription);
- doAnswer(invocation -> {
- dispatcher.consumerFlow(invocation.getArgument(0), invocation.getArgument(1));
- return null;
- }).when(subscription).consumerFlow(any(), anyInt());
-
- Consumer remainingConsumer = createConsumer(subscription, Shared, topicName, 1);
- Consumer removedConsumer = createConsumer(subscription, Shared, topicName, 2);
- dispatcher.addConsumer(remainingConsumer).join();
- dispatcher.addConsumer(removedConsumer).join();
+ try (MockScope mocks = new MockScope()) {
+ String topicName = newTopicName();
+ String subscriptionName = "shared-sub";
+ admin.topics().createNonPartitionedTopic(topicName);
+ PersistentTopic realTopic = (PersistentTopic) getTopic(topicName, false).join().orElseThrow();
+ BrokerService isolatedBroker = mocks.spy(realTopic.getBrokerService());
+ PersistentTopic isolatedTopic = mocks.spy(realTopic);
+ doReturn(isolatedBroker).when(isolatedTopic).getBrokerService();
+
+ EventLoopGroup rejectingExecutor = mocks.mock(EventLoopGroup.class);
+ doThrow(new RejectedExecutionException("test rejection"))
+ .when(rejectingExecutor).execute(any(Runnable.class));
+ when(rejectingExecutor.isShuttingDown()).thenReturn(true);
+ doReturn(rejectingExecutor).when(isolatedBroker).executor();
+
+ ManagedCursor cursor = mocks.mock(ManagedCursorImpl.class);
+ when(cursor.getName()).thenReturn(subscriptionName);
+ when(cursor.isClosed()).thenReturn(true);
+ Subscription subscription = mocks.mock(PersistentSubscription.class);
+ when(subscription.getName()).thenReturn(subscriptionName);
+ when(subscription.getTopic()).thenReturn(isolatedTopic);
+ Dispatcher dispatcher = classic
+ ? new PersistentDispatcherMultipleConsumersClassic(isolatedTopic, cursor, subscription)
+ : new PersistentDispatcherMultipleConsumers(isolatedTopic, cursor, subscription);
+ doAnswer(invocation -> {
+ dispatcher.consumerFlow(invocation.getArgument(0), invocation.getArgument(1));
+ return null;
+ }).when(subscription).consumerFlow(any(), anyInt());
+
+ Consumer remainingConsumer = createConsumer(subscription, Shared, topicName, 1);
+ Consumer removedConsumer = createConsumer(subscription, Shared, topicName, 2);
+ dispatcher.addConsumer(remainingConsumer).join();
+ dispatcher.addConsumer(removedConsumer).join();
- removedConsumer.flowPermits(100);
+ removedConsumer.flowPermits(100);
- assertThat(removedConsumer.getAvailablePermits()).isEqualTo(100);
- assertThat(removedConsumer.getAvailablePermitsForDispatcherRemoval()).isZero();
- assertThat(totalAvailablePermits(dispatcher)).isZero();
- dispatcher.removeConsumer(removedConsumer);
- assertThat(dispatcher.getConsumers()).containsExactly(remainingConsumer);
- assertThat(removedConsumer.getAvailablePermitsForDispatcherRemoval()).isZero();
- assertThat(totalAvailablePermits(dispatcher)).isZero();
- verify(rejectingExecutor).execute(any(Runnable.class));
+ assertThat(removedConsumer.getAvailablePermits()).isEqualTo(100);
+ assertThat(removedConsumer.getAvailablePermitsForDispatcherRemoval()).isZero();
+ assertThat(totalAvailablePermits(dispatcher)).isZero();
+ dispatcher.removeConsumer(removedConsumer);
+ assertThat(dispatcher.getConsumers()).containsExactly(remainingConsumer);
+ assertThat(removedConsumer.getAvailablePermitsForDispatcherRemoval()).isZero();
+ assertThat(totalAvailablePermits(dispatcher)).isZero();
+ verify(rejectingExecutor).execute(any(Runnable.class));
+ }
}
@Test(timeOut = 30_000)
public void testPendingPermitAccountingSurvivesSignedIntegerWrap() throws Exception {
- String topicName = newTopicName();
- String subscriptionName = "shared-sub";
- admin.topics().createNonPartitionedTopic(topicName);
- PersistentTopic topic = (PersistentTopic) getTopic(topicName, false).join().orElseThrow();
- Subscription subscription = mock(PersistentSubscription.class);
- when(subscription.getName()).thenReturn(subscriptionName);
- when(subscription.getTopic()).thenReturn(topic);
- Consumer consumer = createConsumer(subscription, Shared, topicName, 1);
-
- consumer.flowPermits(Integer.MAX_VALUE);
- consumer.flowPermits(Integer.MAX_VALUE);
- consumer.flowPermits(Integer.MAX_VALUE);
-
- assertThat(consumer.getAvailablePermits()).isEqualTo(Integer.MAX_VALUE - 2);
- assertThat(consumer.getAvailablePermitsForDispatcherRemoval()).isZero();
-
- consumer.completePendingDispatcherFlow(Integer.MAX_VALUE);
- assertThat(consumer.getAvailablePermitsForDispatcherRemoval()).isEqualTo(Integer.MAX_VALUE);
- consumer.completePendingDispatcherFlow(Integer.MAX_VALUE);
- assertThat(consumer.getAvailablePermitsForDispatcherRemoval()).isEqualTo(-2);
- consumer.completePendingDispatcherFlow(Integer.MAX_VALUE);
- assertThat(consumer.getAvailablePermitsForDispatcherRemoval()).isEqualTo(Integer.MAX_VALUE - 2);
+ try (MockScope mocks = new MockScope()) {
+ String topicName = newTopicName();
+ String subscriptionName = "shared-sub";
+ admin.topics().createNonPartitionedTopic(topicName);
+ PersistentTopic topic = (PersistentTopic) getTopic(topicName, false).join().orElseThrow();
+ Subscription subscription = mocks.mock(PersistentSubscription.class);
+ when(subscription.getName()).thenReturn(subscriptionName);
+ when(subscription.getTopic()).thenReturn(topic);
+ Consumer consumer = createConsumer(subscription, Shared, topicName, 1);
+
+ consumer.flowPermits(Integer.MAX_VALUE);
+ consumer.flowPermits(Integer.MAX_VALUE);
+ consumer.flowPermits(Integer.MAX_VALUE);
+
+ assertThat(consumer.getAvailablePermits()).isEqualTo(Integer.MAX_VALUE - 2);
+ assertThat(consumer.getAvailablePermitsForDispatcherRemoval()).isZero();
+
+ consumer.completePendingDispatcherFlow(Integer.MAX_VALUE);
+ assertThat(consumer.getAvailablePermitsForDispatcherRemoval()).isEqualTo(Integer.MAX_VALUE);
+ consumer.completePendingDispatcherFlow(Integer.MAX_VALUE);
+ assertThat(consumer.getAvailablePermitsForDispatcherRemoval()).isEqualTo(-2);
+ consumer.completePendingDispatcherFlow(Integer.MAX_VALUE);
+ assertThat(consumer.getAvailablePermitsForDispatcherRemoval()).isEqualTo(Integer.MAX_VALUE - 2);
+ }
}
@Test(dataProvider = "flowRaceDispatcherVariants", timeOut = 30_000)
@@ -419,13 +471,9 @@ public class SharedDispatcherPermitAccountingTest extends SharedPulsarBaseTest {
admin.topics().createNonPartitionedTopic(topicName);
PersistentTopic topic = (PersistentTopic) getTopic(topicName, false).join().get();
- ManagedCursor cursor = mock(ManagedCursorImpl.class);
- when(cursor.getName()).thenReturn(subscriptionName);
- when(cursor.isClosed()).thenReturn(true);
- Subscription subscription = mock(PersistentSubscription.class);
- when(subscription.getName()).thenReturn(subscriptionName);
- when(subscription.getTopic()).thenReturn(topic);
-
+ ManagedCursor cursor = topic.getManagedLedger().newNonDurableCursor(PositionFactory.EARLIEST);
+ cursor.close();
+ PersistentSubscription subscription = new PersistentSubscription(topic, subscriptionName, cursor, false);
Dispatcher dispatcher;
if (subType == Key_Shared) {
dispatcher = classic
@@ -438,10 +486,7 @@ public class SharedDispatcherPermitAccountingTest extends SharedPulsarBaseTest {
? new PersistentDispatcherMultipleConsumersClassic(topic, cursor, subscription)
: new PersistentDispatcherMultipleConsumers(topic, cursor, subscription);
}
- doAnswer(invocation -> {
- dispatcher.consumerFlow(invocation.getArgument(0), invocation.getArgument(1));
- return null;
- }).when(subscription).consumerFlow(any(), anyInt());
+ subscription.dispatcher = dispatcher;
Consumer remainingConsumer = createConsumer(subscription, subType, topicName, 1);
Consumer removedConsumer = createConsumer(subscription, subType, topicName, 2);
@@ -452,27 +497,60 @@ public class SharedDispatcherPermitAccountingTest extends SharedPulsarBaseTest {
}
private Consumer createConsumer(Subscription subscription, SubType subType, String topicName, long consumerId) {
- TransportCnx cnx = mock(TransportCnx.class);
- PulsarCommandSender commandSender = mock(PulsarCommandSender.class);
- when(cnx.isActive()).thenReturn(true);
- when(cnx.isWritable()).thenReturn(true);
- when(cnx.getCommandSender()).thenReturn(commandSender);
- when(commandSender.sendMessagesToConsumer(anyLong(), anyString(), any(), anyInt(), anyList(),
- any(EntryBatchSizes.class), any(EntryBatchIndexesAcks.class), any(RedeliveryTracker.class), anyLong()))
- .thenReturn(ImmediateEventExecutor.INSTANCE.newSucceededFuture(null));
+ ServerCnx cnx = new ServerCnx(subscription.getTopic().getBrokerService().getPulsar()) {
+ @Override
+ public boolean equals(Object other) {
+ return this == other;
+ }
+
+ @Override
+ public int hashCode() {
+ return System.identityHashCode(this);
+ }
+
+ @Override
+ public boolean isActive() {
+ return true;
+ }
+
+ @Override
+ public boolean isWritable() {
+ return true;
+ }
+
+ @Override
+ public String clientSourceAddressAndPort() {
+ return "test";
+ }
+
+ @Override
+ public PulsarCommandSender getCommandSender() {
+ return new PulsarCommandSenderImpl(null, this, null) {
+ @Override
+ public ChannelPromise sendMessagesToConsumer(long id, String name, Subscription sub,
+ int partition, List<? extends Entry> entries, EntryBatchSizes sizes,
+ EntryBatchIndexesAcks indexes, RedeliveryTracker tracker, long epoch) {
+ entries.stream().filter(Objects::nonNull).forEach(Entry::release);
+ sizes.recyle();
+ if (indexes != null) {
+ indexes.recycle();
+ }
+ return new DefaultChannelPromise(channel, ImmediateEventExecutor.INSTANCE).setSuccess();
+ }
+ };
+ }
+ };
return new Consumer(subscription, subType, topicName, consumerId, 0, "consumer-" + consumerId,
true, cnx, "role", emptyMap(), false, new KeySharedMeta().setKeySharedMode(AUTO_SPLIT),
MessageId.latest, DEFAULT_CONSUMER_EPOCH);
}
private static void simulateDispatch(Dispatcher dispatcher, Consumer consumer, int permits) {
- Entry entry = mock(Entry.class);
- when(entry.getLedgerId()).thenReturn(1L);
- when(entry.getEntryId()).thenReturn(1L);
+ Entry entry = EntryImpl.create(1, 1, new byte[0]);
EntryBatchSizes batchSizes = EntryBatchSizes.get(1);
batchSizes.setBatchSize(0, permits);
EntryBatchIndexesAcks batchIndexesAcks = EntryBatchIndexesAcks.get(1);
- RedeliveryTracker redeliveryTracker = mock(RedeliveryTracker.class);
+ RedeliveryTracker redeliveryTracker = RedeliveryTrackerDisabled.REDELIVERY_TRACKER_DISABLED;
List<Integer> stickyKeyHashes = null;
if (dispatcher instanceof StickyKeyDispatcher stickyKeyDispatcher) {
List<Range> ranges = stickyKeyDispatcher.getConsumerKeyHashRanges().get(consumer);
@@ -483,9 +561,6 @@ public class SharedDispatcherPermitAccountingTest extends SharedPulsarBaseTest {
consumer.sendMessages(new ArrayList<>(List.of(entry)), stickyKeyHashes, batchSizes, batchIndexesAcks,
permits, 0, 0, redeliveryTracker, DEFAULT_CONSUMER_EPOCH).syncUninterruptibly();
decrementTotalAvailablePermits(dispatcher, permits);
-
- batchSizes.recyle();
- batchIndexesAcks.recycle();
}
private static void decrementTotalAvailablePermits(Dispatcher dispatcher, int permits) {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/SubscriptionStatsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/SubscriptionStatsTest.java
index ddf99e384f26..6de70dbe0d6f 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/SubscriptionStatsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/SubscriptionStatsTest.java
@@ -18,10 +18,8 @@
*/
package org.apache.pulsar.broker.stats;
-import static org.apache.pulsar.broker.BrokerTestUtil.spyWithClassAndConstructorArgs;
import static org.apache.pulsar.broker.stats.prometheus.PrometheusMetricsClient.Metric;
import static org.apache.pulsar.broker.stats.prometheus.PrometheusMetricsClient.parseMetrics;
-import static org.mockito.Mockito.mock;
import com.google.common.collect.Multimap;
import java.io.ByteArrayOutputStream;
import java.lang.reflect.Field;
@@ -39,14 +37,12 @@ import org.apache.pulsar.broker.service.SharedPulsarBaseTest;
import org.apache.pulsar.broker.service.SharedPulsarCluster;
import org.apache.pulsar.broker.service.plugin.EntryFilter;
import org.apache.pulsar.broker.service.plugin.EntryFilterTest;
-import org.apache.pulsar.broker.service.plugin.EntryFilterWithClassLoader;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.common.naming.TopicName;
-import org.apache.pulsar.common.nar.NarClassLoader;
import org.apache.pulsar.common.policies.data.SubscriptionStats;
import org.apache.pulsar.common.policies.data.TopicStats;
import org.apache.pulsar.common.policies.data.impl.DispatchRateImpl;
@@ -227,11 +223,8 @@ public class SubscriptionStatsTest extends SharedPulsarBaseTest {
field.setAccessible(true);
Field hasFilterField = EntryFilterSupport.class.getDeclaredField("hasFilter");
hasFilterField.setAccessible(true);
- NarClassLoader narClassLoader = mock(NarClassLoader.class);
EntryFilter filter1 = new EntryFilterTest();
- EntryFilterWithClassLoader loader1 =
- spyWithClassAndConstructorArgs(EntryFilterWithClassLoader.class, filter1, narClassLoader, false);
- field.set(dispatcher, List.of(loader1));
+ field.set(dispatcher, List.of(filter1));
hasFilterField.set(dispatcher, true);
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MultiTopicsConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MultiTopicsConsumerTest.java
index 066927365fe9..88d6b4592eb0 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MultiTopicsConsumerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MultiTopicsConsumerTest.java
@@ -18,11 +18,6 @@
*/
package org.apache.pulsar.client.api;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.anyLong;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertTrue;
@@ -40,7 +35,8 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
@@ -58,12 +54,11 @@ import org.apache.pulsar.client.impl.MessageIdImpl;
import org.apache.pulsar.client.impl.MultiTopicsConsumerImpl;
import org.apache.pulsar.client.impl.PulsarClientImpl;
import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.apache.pulsar.client.util.ScheduledExecutorProvider;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.utils.TestLogAppender;
import org.awaitility.Awaitility;
-import org.mockito.AdditionalAnswers;
-import org.mockito.Mockito;
import org.testng.Assert;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.DataProvider;
@@ -72,8 +67,6 @@ import org.testng.annotations.Test;
@CustomLog
@Test(groups = "broker")
public class MultiTopicsConsumerTest extends SharedPulsarBaseTest {
- private ScheduledExecutorService internalExecutorServiceDelegate;
-
protected String methodName;
@BeforeMethod(alwaysRun = true)
@@ -90,24 +83,23 @@ public class MultiTopicsConsumerTest extends SharedPulsarBaseTest {
ClientConfigurationData conf = ((ClientBuilderImpl) PulsarClient.builder().serviceUrl(getBrokerServiceUrl()))
.getClientConfigurationData();
- @Cleanup
- PulsarClientImpl client = new PulsarClientImpl(conf) {
- {
- ScheduledExecutorService internalExecutorService =
- (ScheduledExecutorService) super.getScheduledExecutorProvider().getExecutor();
- internalExecutorServiceDelegate = mock(ScheduledExecutorService.class,
- // a spy isn't used since that doesn't work for private classes, instead
- // the mock delegatesTo an existing instance. A delegate is sufficient for verifying
- // method calls on the interface.
- Mockito.withSettings().defaultAnswer(AdditionalAnswers.delegatesTo(internalExecutorService)));
- }
-
+ AtomicInteger schedulingOperations = new AtomicInteger();
+ @Cleanup("shutdownNow")
+ ScheduledExecutorProvider scheduledExecutors = new ScheduledExecutorProvider(1, "multi-topics-test") {
@Override
- public ExecutorService getInternalExecutorService() {
- return internalExecutorServiceDelegate;
+ protected ExecutorService createExecutor(ExtendedThreadFactory threadFactory) {
+ return new ScheduledThreadPoolExecutor(1, threadFactory) {
+ @Override
+ public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
+ schedulingOperations.incrementAndGet();
+ return super.schedule(command, delay, unit);
+ }
+ };
}
};
@Cleanup
+ PulsarClientImpl client = new PulsarClientImpl(conf, null, null, null, null, null, scheduledExecutors);
+ @Cleanup
Producer<byte[]> producer1 = client.newProducer()
.topic(topicNameBase + "1")
.enableBatching(false)
@@ -140,8 +132,7 @@ public class MultiTopicsConsumerTest extends SharedPulsarBaseTest {
Thread.sleep(1000L);
// then verify that no scheduling operation has happened
- verify(internalExecutorServiceDelegate, times(0))
- .schedule(any(Runnable.class), anyLong(), any());
+ assertEquals(schedulingOperations.get(), 0);
}
@SuppressWarnings({"deprecation", "unchecked"})
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerReconnectionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerReconnectionTest.java
index dddf51a97b79..6553e544171d 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerReconnectionTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerReconnectionTest.java
@@ -18,8 +18,6 @@
*/
package org.apache.pulsar.client.impl;
-import static org.mockito.Mockito.doAnswer;
-import static org.mockito.Mockito.spy;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotEquals;
import static org.testng.Assert.assertTrue;
@@ -60,18 +58,25 @@ public class ProducerReconnectionTest extends SharedPulsarBaseTest {
@Override
ConnectionHandler initConnectionHandler() {
ConnectionHandler connectionHandler = super.initConnectionHandler();
- ConnectionHandler spyConnectionHandler = spy(connectionHandler);
- doAnswer(invocation -> {
- boolean result = (boolean) invocation.callRealMethod();
- if (reconnectionStartTrigger.get()) {
- log.info("[testConcurrencyReconnectAndClose] verified state for reconnection");
- reconnectingSignal.countDown();
- closedSignal.await();
- log.info("[testConcurrencyReconnectAndClose] reconnected");
+ return new ConnectionHandler(connectionHandler.state, connectionHandler.backoff,
+ connectionHandler.connection) {
+ @Override
+ public boolean isValidStateForReconnection() {
+ boolean result = super.isValidStateForReconnection();
+ if (reconnectionStartTrigger.get()) {
+ reconnectingSignal.countDown();
+ try {
+ if (!closedSignal.await(10, TimeUnit.SECONDS)) {
+ throw new IllegalStateException("Producer close did not release reconnection");
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(e);
+ }
+ }
+ return result;
}
- return result;
- }).when(spyConnectionHandler).isValidStateForReconnection();
- return spyConnectionHandler;
+ };
}
};
log.info("[testConcurrencyReconnectAndClose] producer created");
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/SimpleProduceConsumeIoTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/SimpleProduceConsumeIoTest.java
index fca8f55e2bf9..40a97e4ec9cc 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/SimpleProduceConsumeIoTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/SimpleProduceConsumeIoTest.java
@@ -18,14 +18,11 @@
*/
package org.apache.pulsar.client.impl;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.doAnswer;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.spy;
-import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
-import io.netty.channel.ChannelFuture;
-import io.netty.util.concurrent.GenericFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelOutboundHandlerAdapter;
+import io.netty.channel.ChannelPromise;
+import io.netty.util.ReferenceCountUtil;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
@@ -149,25 +146,15 @@ public class SimpleProduceConsumeIoTest extends SharedPulsarBaseTest {
}
private void setFailedContext(ClientCnx cnx) {
- final var oldCtx = cnx.ctx();
- final var newCtx = spy(oldCtx);
- doAnswer(invocationOnMock -> {
- final var failedFuture = mock(ChannelFuture.class);
- doAnswer(invocation -> {
- @SuppressWarnings("unchecked")
- final var listener = (GenericFutureListener<ChannelFuture>) invocation.getArgument(0);
- final var future = mock(ChannelFuture.class);
- when(future.isSuccess()).thenReturn(false);
- when(future.cause()).thenReturn(new RuntimeException("network exception"));
- listener.operationComplete(future);
- return future;
- }).when(failedFuture).addListener(any());
- // Set back the original context because reconnection will still get the same `ClientCnx` from the pool
- cnx.setCtx(oldCtx);
- return failedFuture;
- }).when(newCtx).writeAndFlush(any());
-
- cnx.setCtx(newCtx);
+ ChannelHandlerContext context = cnx.ctx();
+ context.pipeline().addBefore(context.name(), "fail-next-write", new ChannelOutboundHandlerAdapter() {
+ @Override
+ public void write(ChannelHandlerContext ctx, Object message, ChannelPromise promise) {
+ ctx.pipeline().remove(this);
+ ReferenceCountUtil.release(message);
+ promise.setFailure(new RuntimeException("network exception"));
+ }
+ });
}
@RequiredArgsConstructor
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/stats/client/PulsarBrokerStatsClientTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/stats/client/PulsarBrokerStatsClientTest.java
index b17e874ace7a..9376983c93a6 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/stats/client/PulsarBrokerStatsClientTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/stats/client/PulsarBrokerStatsClientTest.java
@@ -18,7 +18,6 @@
*/
package org.apache.pulsar.stats.client;
-import static org.mockito.Mockito.spy;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotNull;
@@ -68,7 +67,7 @@ public class PulsarBrokerStatsClientTest extends SharedPulsarBaseTest {
URL url = new URL("http://localhost:15000");
@Cleanup
PulsarAdmin admin = PulsarAdmin.builder().serviceHttpUrl(url.toString()).build();
- BrokerStatsImpl client = (BrokerStatsImpl) spy(admin.brokerStats());
+ BrokerStatsImpl client = (BrokerStatsImpl) admin.brokerStats();
try {
client.getLoadReport();
} catch (PulsarAdminException e) {
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/netty/ChannelFuturesTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/netty/ChannelFuturesTest.java
index 65b9e7b6339e..15d634bc5977 100644
--- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/netty/ChannelFuturesTest.java
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/netty/ChannelFuturesTest.java
@@ -18,17 +18,16 @@
*/
package org.apache.pulsar.common.util.netty;
-import static org.mockito.Mockito.when;
import io.netty.channel.Channel;
import io.netty.channel.DefaultChannelPromise;
import io.netty.channel.DefaultEventLoop;
+import io.netty.channel.local.LocalChannel;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
-import org.mockito.Mock;
-import org.mockito.MockitoAnnotations;
import org.testng.Assert;
+import org.testng.annotations.AfterMethod;
import org.testng.annotations.AfterTest;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.BeforeTest;
@@ -39,7 +38,6 @@ import org.testng.annotations.Test;
*/
public class ChannelFuturesTest {
- @Mock
private Channel channel;
private DefaultEventLoop eventLoop;
@@ -54,18 +52,26 @@ public class ChannelFuturesTest {
@AfterTest(alwaysRun = true)
public void shutdownEventLoop() throws InterruptedException {
if (eventLoop != null) {
- eventLoop.shutdownGracefully(0, 0, TimeUnit.MILLISECONDS).await(100);
+ eventLoop.shutdownGracefully(0, 0, TimeUnit.MILLISECONDS).sync();
}
}
@BeforeMethod
- public void setup() {
- MockitoAnnotations.openMocks(this);
- when(channel.eventLoop()).thenReturn(eventLoop);
-
+ public void setup() throws InterruptedException {
+ channel = new LocalChannel();
+ eventLoop.register(channel).sync();
channelFuture = new DefaultChannelPromise(channel);
}
+ @AfterMethod(alwaysRun = true)
+ public void closeChannel() throws InterruptedException {
+ if (channel != null) {
+ channel.close().sync();
+ channelFuture = null;
+ channel = null;
+ }
+ }
+
@Test
public void toCompletableFuture_shouldRequireNonNullArgument() {
CompletableFuture<Channel> future = ChannelFutures.toCompletableFuture(null);
diff --git a/pulsar-websocket/src/test/java/org/apache/pulsar/websocket/admin/WebSocketWebResourceTest.java b/pulsar-websocket/src/test/java/org/apache/pulsar/websocket/admin/WebSocketWebResourceTest.java
index 123bba0d22b4..a6e4c73e0c7e 100644
--- a/pulsar-websocket/src/test/java/org/apache/pulsar/websocket/admin/WebSocketWebResourceTest.java
+++ b/pulsar-websocket/src/test/java/org/apache/pulsar/websocket/admin/WebSocketWebResourceTest.java
@@ -22,7 +22,6 @@ import static org.mockito.Mockito.any;
import static org.mockito.Mockito.anyString;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.eq;
-import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import com.google.common.collect.Sets;
import jakarta.servlet.ServletContext;
@@ -30,6 +29,8 @@ import jakarta.servlet.http.HttpServletRequest;
import jakarta.ws.rs.core.Response.Status;
import jakarta.ws.rs.core.UriInfo;
import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.List;
import javax.naming.AuthenticationException;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.authentication.AuthenticationDataHttps;
@@ -41,6 +42,7 @@ import org.apache.pulsar.common.util.RestException;
import org.apache.pulsar.websocket.WebSocketService;
import org.mockito.InjectMocks;
import org.mockito.Mock;
+import org.mockito.Mockito;
import org.mockito.MockitoAnnotations;
import org.mockito.Spy;
import org.testng.Assert;
@@ -55,6 +57,8 @@ public class WebSocketWebResourceTest {
private static final String UNAUTHORIZED_USER = "unauthorized";
private TopicName topicName;
+ private AutoCloseable mockAnnotations;
+ private final List<Object> ownedMocks = new ArrayList<>();
@InjectMocks
@Spy
@@ -69,7 +73,11 @@ public class WebSocketWebResourceTest {
@SuppressWarnings("deprecation")
@BeforeMethod
public void setup(Method method) throws Exception {
- MockitoAnnotations.openMocks(this);
+ mockAnnotations = MockitoAnnotations.openMocks(this);
+ ownedMocks.add(webResource);
+ ownedMocks.add(servletContext);
+ ownedMocks.add(httpRequest);
+ ownedMocks.add(uri);
ServiceConfiguration config = new ServiceConfiguration();
config.setSuperUserRoles(Sets.newHashSet(SUPER_USER));
@@ -129,7 +137,30 @@ public class WebSocketWebResourceTest {
@AfterMethod(alwaysRun = true)
public void cleanup() throws Exception {
- this.webResource = null;
+ try {
+ Mockito.reset(ownedMocks.toArray());
+ } finally {
+ try {
+ ownedMocks.forEach(mock -> Mockito.framework().clearInlineMock(mock));
+ if (mockAnnotations != null) {
+ mockAnnotations.close();
+ }
+ } finally {
+ ownedMocks.clear();
+ mockAnnotations = null;
+ webResource = null;
+ servletContext = null;
+ httpRequest = null;
+ uri = null;
+ topicName = null;
+ }
+ }
+ }
+
+ private <T> T mock(Class<T> type) {
+ T mock = Mockito.mock(type);
+ ownedMocks.add(mock);
+ return mock;
}
@Test