diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/TestPulsarService.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/TestPulsarService.java deleted file mode 100644 index 95667f95a335d..0000000000000 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/TestPulsarService.java +++ /dev/null @@ -1,501 +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.broker; - -import static org.apache.pulsar.broker.BrokerTestUtil.spyWithClassAndConstructorArgs; -import static org.apache.pulsar.broker.BrokerTestUtil.spyWithClassAndConstructorArgsRecordingInvocations; -import static org.mockito.Mockito.mock; -import io.netty.channel.EventLoopGroup; -import io.netty.channel.nio.NioEventLoopGroup; -import java.io.IOException; -import java.time.Duration; -import java.util.Collections; -import java.util.List; -import java.util.Map; -import java.util.Optional; -import java.util.concurrent.CompletableFuture; -import java.util.function.Function; -import java.util.function.Supplier; -import lombok.AccessLevel; -import lombok.Builder; -import lombok.Getter; -import lombok.Singular; -import lombok.ToString; -import lombok.extern.slf4j.Slf4j; -import org.apache.bookkeeper.client.BookKeeper; -import org.apache.bookkeeper.client.PulsarMockBookKeeper; -import org.apache.bookkeeper.common.util.OrderedExecutor; -import org.apache.bookkeeper.mledger.ManagedLedgerFactory; -import org.apache.bookkeeper.stats.NullStatsProvider; -import org.apache.bookkeeper.stats.StatsProvider; -import org.apache.pulsar.broker.namespace.NamespaceService; -import org.apache.pulsar.broker.resources.NamespaceResources; -import org.apache.pulsar.broker.resources.PulsarResources; -import org.apache.pulsar.broker.resources.TopicResources; -import org.apache.pulsar.broker.service.BrokerService; -import org.apache.pulsar.broker.service.PulsarMetadataEventSynchronizer; -import org.apache.pulsar.broker.service.ServerCnx; -import org.apache.pulsar.broker.service.schema.DefaultSchemaRegistryService; -import org.apache.pulsar.broker.service.schema.SchemaRegistryService; -import org.apache.pulsar.broker.storage.ManagedLedgerStorage; -import org.apache.pulsar.client.api.PulsarClient; -import org.apache.pulsar.client.impl.PulsarClientImpl; -import org.apache.pulsar.common.naming.TopicName; -import org.apache.pulsar.common.util.GracefulExecutorServicesShutdown; -import org.apache.pulsar.compaction.Compactor; -import org.apache.pulsar.metadata.api.MetadataStore; -import org.apache.pulsar.metadata.api.MetadataStoreConfig; -import org.apache.pulsar.metadata.api.MetadataStoreException; -import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended; -import org.apache.pulsar.metadata.impl.MetadataStoreFactoryImpl; -import org.jetbrains.annotations.NotNull; - -/** - * Subclass of PulsarService that is used for some tests. - * This was written as a replacement for the previous Mockito Spy over PulsarService solution which caused - * a flaky test issue https://github.com/apache/pulsar/issues/13620. - */ - -public class TestPulsarService extends PulsarService { - - - @Slf4j - @ToString - @Getter - @Builder - public static class Factory implements AutoCloseable { - private final ServiceConfiguration config; - private final MetadataStoreExtended localMetadataStore; - private final MetadataStoreExtended configurationMetadataStore; - private final PulsarResources pulsarResources; - - private final OrderedExecutor executor; - - private final EventLoopGroup eventLoopGroup; - - private final ManagedLedgerStorage managedLedgerClientFactory; - - private final boolean useSpies; - - private final PulsarService pulsarService; - - private final Compactor compactor; - - private final BrokerService brokerService; - - @Getter(AccessLevel.NONE) - @Singular - private final List cleanupFunctions; - - public ManagedLedgerFactory getManagedLedgerFactory() { - return managedLedgerClientFactory.getManagedLedgerFactory(); - } - - public static FactoryBuilder builder() { - return new CustomFactoryBuilder(); - } - - public void close() throws Exception { - pulsarService.getBrokerService().close(); - pulsarService.close(); - GracefulExecutorServicesShutdown.initiate() - .timeout(Duration.ZERO) - .shutdown(executor) - .handle().get(); - eventLoopGroup.shutdownGracefully().get(); - if (localMetadataStore != configurationMetadataStore) { - localMetadataStore.close(); - configurationMetadataStore.close(); - } else { - localMetadataStore.close(); - } - for (AutoCloseable cleanup : cleanupFunctions) { - try { - cleanup.close(); - } catch (Exception e) { - log.error("Failure in calling cleanup function", e); - } - } - } - - public ServerCnx createServerCnxSpy() { - return spyWithClassAndConstructorArgsRecordingInvocations(ServerCnx.class, - getPulsarService()); - } - - public static class FactoryBuilder { - protected boolean useTestPulsarResources = false; - protected MetadataStore pulsarResourcesMetadataStore; - protected Function brokerServiceFunction; - - public FactoryBuilder useTestPulsarResources() { - useTestPulsarResources = true; - return this; - } - - public FactoryBuilder useTestPulsarResources(MetadataStore metadataStore) { - useTestPulsarResources = true; - pulsarResourcesMetadataStore = metadataStore; - return this; - } - - public FactoryBuilder managedLedgerClients(BookKeeper bookKeeperClient, - ManagedLedgerFactory managedLedgerFactory) { - return managedLedgerClientFactory( - Factory.createManagedLedgerClientFactory(bookKeeperClient, managedLedgerFactory)); - } - - public FactoryBuilder brokerServiceFunction( - Function brokerServiceFunction) { - this.brokerServiceFunction = brokerServiceFunction; - return this; - } - } - - private static class CustomFactoryBuilder extends Factory.FactoryBuilder { - - @Override - public Factory build() { - initializeDefaults(); - return super.build(); - } - - private void initializeDefaults() { - try { - if (super.managedLedgerClientFactory == null) { - if (super.executor == null) { - super.executor = OrderedExecutor.newBuilder().numThreads(1) - .name(TestPulsarService.class.getSimpleName() + "-executor").build(); - } - NonClosableMockBookKeeper mockBookKeeper; - if (super.useSpies) { - mockBookKeeper = - spyWithClassAndConstructorArgs(NonClosableMockBookKeeper.class, super.executor); - } else { - mockBookKeeper = new NonClosableMockBookKeeper(super.executor); - } - cleanupFunction(() -> mockBookKeeper.reallyShutdown()); - ManagedLedgerFactory mlFactoryMock = mock(ManagedLedgerFactory.class); - - managedLedgerClientFactory( - Factory.createManagedLedgerClientFactory(mockBookKeeper, mlFactoryMock)); - } - } catch (Exception e) { - throw new RuntimeException(e); - } - if (super.config == null) { - ServiceConfiguration svcConfig = new ServiceConfiguration(); - svcConfig.setBrokerShutdownTimeoutMs(0L); - svcConfig.setLoadBalancerOverrideBrokerNicSpeedGbps(Optional.of(1.0d)); - svcConfig.setClusterName("pulsar-cluster"); - config(svcConfig); - } - if (super.localMetadataStore == null || super.configurationMetadataStore == null) { - try { - MetadataStoreExtended store = MetadataStoreFactoryImpl.createExtended("memory:local", - MetadataStoreConfig.builder().build()); - if (super.localMetadataStore == null) { - localMetadataStore(store); - } - if (super.configurationMetadataStore == null) { - configurationMetadataStore(store); - } - } catch (MetadataStoreException e) { - throw new RuntimeException(e); - } - } - if (super.pulsarResources == null) { - if (useTestPulsarResources) { - MetadataStore metadataStore = pulsarResourcesMetadataStore; - if (metadataStore == null) { - metadataStore = super.configurationMetadataStore; - } - NamespaceResources nsr = - spyWithClassAndConstructorArgs(NamespaceResources.class, metadataStore, 30); - TopicResources tsr = spyWithClassAndConstructorArgs(TopicResources.class, metadataStore); - if (!super.useSpies) { - pulsarResources( - new TestPulsarResources(super.localMetadataStore, super.configurationMetadataStore, - tsr, nsr)); - } else { - pulsarResources( - spyWithClassAndConstructorArgs(TestPulsarResources.class, super.localMetadataStore, - super.configurationMetadataStore, tsr, nsr)); - } - } else { - if (!super.useSpies) { - pulsarResources( - new PulsarResources(super.localMetadataStore, super.configurationMetadataStore)); - } else { - pulsarResources( - spyWithClassAndConstructorArgs(PulsarResources.class, super.localMetadataStore, - super.configurationMetadataStore)); - } - } - } - if (super.brokerServiceFunction == null) { - if (super.brokerService == null) { - if (super.eventLoopGroup == null) { - super.eventLoopGroup = new NioEventLoopGroup(); - } - brokerServiceFunction(pulsarService -> { - try { - if (!super.useSpies) { - return new TestBrokerService(pulsarService, super.eventLoopGroup); - } else { - return spyWithClassAndConstructorArgs(TestBrokerService.class, pulsarService, - super.eventLoopGroup); - } - } catch (Exception e) { - throw new RuntimeException(e); - } - }); - } else { - brokerServiceFunction(pulsarService -> super.brokerService); - } - } - if (!super.useSpies) { - pulsarService(new TestPulsarService(super.config, super.localMetadataStore, - super.configurationMetadataStore, super.pulsarResources, super.managedLedgerClientFactory, - super.brokerServiceFunction, super.executor, super.compactor)); - } else { - pulsarService(spyWithClassAndConstructorArgs(TestPulsarService.class, super.config, - super.localMetadataStore, super.configurationMetadataStore, super.pulsarResources, - super.managedLedgerClientFactory, super.brokerServiceFunction, super.executor, - super.compactor)); - } - if (super.brokerService == null) { - brokerService(super.pulsarService.getBrokerService()); - } - } - } - - @NotNull - private static ManagedLedgerStorage createManagedLedgerClientFactory(BookKeeper bookKeeperClient, - ManagedLedgerFactory managedLedgerFactory) { - return new ManagedLedgerStorage() { - - @Override - public void initialize(ServiceConfiguration conf, MetadataStoreExtended metadataStore, - BookKeeperClientFactory bookkeeperProvider, EventLoopGroup eventLoopGroup) - throws Exception { - - } - - @Override - public ManagedLedgerFactory getManagedLedgerFactory() { - return managedLedgerFactory; - } - - @Override - public StatsProvider getStatsProvider() { - return new NullStatsProvider(); - } - - @Override - public BookKeeper getBookKeeperClient() { - return bookKeeperClient; - } - - @Override - public void close() throws IOException { - - } - }; - } - } - - private static class TestPulsarResources extends PulsarResources { - - private final TopicResources topicResources; - private final NamespaceResources namespaceResources; - - public TestPulsarResources(MetadataStore localMetadataStore, MetadataStore configurationMetadataStore, - TopicResources topicResources, NamespaceResources namespaceResources) { - super(localMetadataStore, configurationMetadataStore); - this.topicResources = topicResources; - this.namespaceResources = namespaceResources; - } - - @Override - public TopicResources getTopicResources() { - return topicResources; - } - - @Override - public NamespaceResources getNamespaceResources() { - return namespaceResources; - } - } - - private static class TestBrokerService extends BrokerService { - - TestBrokerService(PulsarService pulsar, EventLoopGroup eventLoopGroup) throws Exception { - super(pulsar, eventLoopGroup); - } - - @Override - protected CompletableFuture> fetchTopicPropertiesAsync(TopicName topicName) { - return CompletableFuture.completedFuture(Collections.emptyMap()); - } - } - - // Prevent the MockBookKeeper instance from being closed when the broker is restarted within a test - private static class NonClosableMockBookKeeper extends PulsarMockBookKeeper { - - public NonClosableMockBookKeeper(OrderedExecutor executor) throws Exception { - super(executor); - } - - @Override - public void close() { - // no-op - } - - @Override - public void shutdown() { - // no-op - } - - public void reallyShutdown() { - super.shutdown(); - } - } - - private final MetadataStoreExtended localMetadataStore; - private final MetadataStoreExtended configurationMetadataStore; - private final PulsarResources pulsarResources; - private final ManagedLedgerStorage managedLedgerClientFactory; - private final BrokerService brokerService; - - private final OrderedExecutor executor; - - private final Compactor compactor; - - private final SchemaRegistryService schemaRegistryService; - - private final PulsarClient pulsarClient; - - private final NamespaceService namespaceService; - - protected TestPulsarService(ServiceConfiguration config, MetadataStoreExtended localMetadataStore, - MetadataStoreExtended configurationMetadataStore, PulsarResources pulsarResources, - ManagedLedgerStorage managedLedgerClientFactory, - Function brokerServiceFunction, OrderedExecutor executor, - Compactor compactor) { - super(config); - this.localMetadataStore = localMetadataStore; - this.configurationMetadataStore = configurationMetadataStore; - this.pulsarResources = pulsarResources; - this.managedLedgerClientFactory = managedLedgerClientFactory; - this.brokerService = brokerServiceFunction.apply(this); - this.executor = executor; - this.compactor = compactor; - this.schemaRegistryService = spyWithClassAndConstructorArgs(DefaultSchemaRegistryService.class); - this.pulsarClient = mock(PulsarClientImpl.class); - this.namespaceService = mock(NamespaceService.class); - try { - startNamespaceService(); - } catch (PulsarServerException e) { - throw new RuntimeException(e); - } - } - - - @Override - public Supplier getNamespaceServiceProvider() throws PulsarServerException { - return () -> namespaceService; - } - - @Override - public synchronized PulsarClient getClient() throws PulsarServerException { - return pulsarClient; - } - - @Override - public SchemaRegistryService getSchemaRegistryService() { - return schemaRegistryService; - } - - @Override - public MetadataStore createConfigurationMetadataStore(PulsarMetadataEventSynchronizer synchronizer) - throws MetadataStoreException { - return configurationMetadataStore; - } - - @Override - public MetadataStoreExtended createLocalMetadataStore(PulsarMetadataEventSynchronizer synchronizer) - throws MetadataStoreException, PulsarServerException { - return localMetadataStore; - } - - @Override - public PulsarResources getPulsarResources() { - return pulsarResources; - } - - public BrokerService getBrokerService() { - return brokerService; - } - - @Override - public Compactor getCompactor() throws PulsarServerException { - if (compactor != null) { - return compactor; - } else { - return super.getCompactor(); - } - } - - @Override - public MetadataStore getConfigurationMetadataStore() { - return configurationMetadataStore; - } - - @Override - public MetadataStoreExtended getLocalMetadataStore() { - return localMetadataStore; - } - - @Override - public ManagedLedgerStorage getManagedLedgerClientFactory() { - return managedLedgerClientFactory; - } - - @Override - public OrderedExecutor getOrderedExecutor() { - return executor; - } - - @Override - protected PulsarResources newPulsarResources() { - return pulsarResources; - } - - @Override - protected ManagedLedgerStorage newManagedLedgerClientFactory() throws Exception { - return managedLedgerClientFactory; - } - - @Override - protected BrokerService newBrokerService(PulsarService pulsar) throws Exception { - return brokerService; - } -} diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/OwnerShipForCurrentServerTestBase.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/OwnerShipForCurrentServerTestBase.java index e4331d0b48627..d5ea10da0f48d 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/OwnerShipForCurrentServerTestBase.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/OwnerShipForCurrentServerTestBase.java @@ -16,54 +16,31 @@ * specific language governing permissions and limitations * under the License. */ + package org.apache.pulsar.broker.namespace; -import static org.apache.pulsar.broker.BrokerTestUtil.spyWithClassAndConstructorArgs; -import static org.mockito.Mockito.doAnswer; -import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.spy; -import com.google.common.util.concurrent.MoreExecutors; -import io.netty.channel.EventLoopGroup; -import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.List; -import java.util.Map; import java.util.Optional; -import java.util.concurrent.TimeUnit; -import java.util.function.Supplier; import lombok.Getter; import lombok.Setter; import lombok.extern.slf4j.Slf4j; -import org.apache.bookkeeper.client.BookKeeper; -import org.apache.bookkeeper.client.EnsemblePlacementPolicy; -import org.apache.bookkeeper.client.PulsarMockBookKeeper; -import org.apache.bookkeeper.common.util.OrderedExecutor; -import org.apache.bookkeeper.stats.StatsLogger; -import org.apache.bookkeeper.util.ZkUtils; -import org.apache.pulsar.broker.BookKeeperClientFactory; import org.apache.pulsar.broker.PulsarService; import org.apache.pulsar.broker.ServiceConfiguration; -import org.apache.pulsar.broker.auth.SameThreadOrderedSafeExecutor; -import org.apache.pulsar.broker.intercept.CounterBrokerInterceptor; +import org.apache.pulsar.broker.testcontext.PulsarTestContext; import org.apache.pulsar.client.admin.PulsarAdmin; import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.common.policies.data.TopicType; -import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended; -import org.apache.pulsar.metadata.impl.ZKMetadataStore; -import org.apache.zookeeper.CreateMode; -import org.apache.zookeeper.MockZooKeeper; -import org.apache.zookeeper.MockZooKeeperSession; -import org.apache.zookeeper.data.ACL; @Slf4j -public class OwnerShipForCurrentServerTestBase { +public abstract class OwnerShipForCurrentServerTestBase { public static final String CLUSTER_NAME = "test"; @Setter private int brokerCount = 3; - private final List orderedExecutorList = new ArrayList<>(); @Getter private final List serviceConfigurationList = new ArrayList<>(); @Getter @@ -72,9 +49,7 @@ public class OwnerShipForCurrentServerTestBase { protected PulsarAdmin admin; protected PulsarClient pulsarClient; - private MockZooKeeper mockZooKeeper; - private OrderedExecutor bkExecutor; - private NonClosableMockBookKeeper mockBookKeeper; + protected List pulsarTestContexts = new ArrayList<>(); public void internalSetup() throws Exception { init(); @@ -85,13 +60,6 @@ public void internalSetup() throws Exception { } private void init() throws Exception { - mockZooKeeper = createMockZooKeeper(); - - bkExecutor = OrderedExecutor.newBuilder() - .numThreads(1) - .name("mock-pulsar-bk") - .build(); - mockBookKeeper = createMockBookKeeper(bkExecutor); startBroker(); } @@ -118,96 +86,22 @@ protected void startBroker() throws Exception { conf.setWebServicePortTls(Optional.of(0)); serviceConfigurationList.add(conf); - PulsarService pulsar = spyWithClassAndConstructorArgs(PulsarService.class, conf); - - setupBrokerMocks(pulsar); - pulsar.start(); + PulsarTestContext.Builder testContextBuilder = + PulsarTestContext.startableBuilder() + .config(conf); + if (i > 0) { + testContextBuilder.reuseMockBookkeeperAndMetadataStores(pulsarTestContexts.get(0)); + } else { + testContextBuilder.withMockZookeeper(); + } + PulsarTestContext pulsarTestContext = testContextBuilder + .build(); + PulsarService pulsar = pulsarTestContext.getPulsarService(); pulsarServiceList.add(pulsar); + pulsarTestContexts.add(pulsarTestContext); } } - protected void setupBrokerMocks(PulsarService pulsar) throws Exception { - // Override default providers with mocked ones - doReturn(mockBookKeeperClientFactory).when(pulsar).newBookKeeperClientFactory(); - MockZooKeeperSession mockZooKeeperSession = MockZooKeeperSession.newInstance(mockZooKeeper); - doReturn(new ZKMetadataStore(mockZooKeeperSession)).when(pulsar).createLocalMetadataStore(null); - doReturn(new ZKMetadataStore(mockZooKeeperSession)).when(pulsar).createConfigurationMetadataStore(null); - Supplier namespaceServiceSupplier = () -> spyWithClassAndConstructorArgs( - NamespaceService.class, pulsar); - doReturn(namespaceServiceSupplier).when(pulsar).getNamespaceServiceProvider(); - - SameThreadOrderedSafeExecutor executor = new SameThreadOrderedSafeExecutor(); - orderedExecutorList.add(executor); - doReturn(executor).when(pulsar).getOrderedExecutor(); - doReturn(new CounterBrokerInterceptor()).when(pulsar).getBrokerInterceptor(); - - doAnswer((invocation) -> spy(invocation.callRealMethod())).when(pulsar).newCompactor(); - } - - public static MockZooKeeper createMockZooKeeper() throws Exception { - MockZooKeeper zk = MockZooKeeper.newInstance(MoreExecutors.newDirectExecutorService()); - List dummyAclList = new ArrayList<>(0); - - ZkUtils.createFullPathOptimistic(zk, "/ledgers/available/192.168.1.1:" + 5000, - "".getBytes(StandardCharsets.UTF_8), dummyAclList, CreateMode.PERSISTENT); - - zk.create("/ledgers/LAYOUT", "1\nflat:1".getBytes(StandardCharsets.UTF_8), dummyAclList, - CreateMode.PERSISTENT); - return zk; - } - - public static NonClosableMockBookKeeper createMockBookKeeper(OrderedExecutor executor) throws Exception { - return spyWithClassAndConstructorArgs(NonClosableMockBookKeeper.class, executor); - } - - // Prevent the MockBookKeeper instance from being closed when the broker is restarted within a test - public static class NonClosableMockBookKeeper extends PulsarMockBookKeeper { - - public NonClosableMockBookKeeper(OrderedExecutor executor) throws Exception { - super(executor); - } - - @Override - public void close() { - // no-op - } - - @Override - public void shutdown() { - // no-op - } - - public void reallyShutdown() { - super.shutdown(); - } - } - - private final BookKeeperClientFactory mockBookKeeperClientFactory = new BookKeeperClientFactory() { - - @Override - public BookKeeper create(ServiceConfiguration conf, MetadataStoreExtended store, - EventLoopGroup eventLoopGroup, - Optional> ensemblePlacementPolicyClass, - Map properties) { - // Always return the same instance (so that we don't loose the mock BK content on broker restart - return mockBookKeeper; - } - - @Override - public BookKeeper create(ServiceConfiguration conf, MetadataStoreExtended store, - EventLoopGroup eventLoopGroup, - Optional> ensemblePlacementPolicyClass, - Map properties, StatsLogger statsLogger) { - // Always return the same instance (so that we don't loose the mock BK content on broker restart - return mockBookKeeper; - } - - @Override - public void close() { - // no-op - } - }; - protected final void internalCleanup() { try { // if init fails, some of these could be null, and if so would throw @@ -220,49 +114,19 @@ protected final void internalCleanup() { pulsarClient.shutdown(); pulsarClient = null; } - if (pulsarServiceList.size() > 0) { - for (PulsarService pulsarService : pulsarServiceList) { - pulsarService.close(); + if (pulsarTestContexts.size() > 0) { + for(int i = pulsarTestContexts.size() - 1; i >= 0; i--) { + pulsarTestContexts.get(i).close(); } - pulsarServiceList.clear(); + pulsarTestContexts.clear(); } + pulsarServiceList.clear(); if (serviceConfigurationList.size() > 0) { serviceConfigurationList.clear(); } - if (mockBookKeeper != null) { - mockBookKeeper.reallyShutdown(); - } - if (mockZooKeeper != null) { - mockZooKeeper.shutdown(); - } - if (orderedExecutorList.size() > 0) { - for (int i = 0; i < orderedExecutorList.size(); i++) { - SameThreadOrderedSafeExecutor sameThreadOrderedSafeExecutor = orderedExecutorList.get(i); - if(sameThreadOrderedSafeExecutor != null) { - try { - sameThreadOrderedSafeExecutor.shutdownNow(); - sameThreadOrderedSafeExecutor.awaitTermination(5, TimeUnit.SECONDS); - } catch (InterruptedException ex) { - log.error("sameThreadOrderedSafeExecutor shutdown had error", ex); - Thread.currentThread().interrupt(); - } - orderedExecutorList.set(i, null); - } - } - } - if(bkExecutor != null) { - try { - bkExecutor.shutdownNow(); - bkExecutor.awaitTermination(5, TimeUnit.SECONDS); - } catch (InterruptedException ex) { - log.error("bkExecutor shutdown had error", ex); - Thread.currentThread().interrupt(); - } - bkExecutor = null; - } + } catch (Exception e) { log.warn("Failed to clean up mocked pulsar service:", e); } } - } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/MessageCumulativeAckTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/MessageCumulativeAckTest.java index 703376b254669..d01fa1fa54028 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/MessageCumulativeAckTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/MessageCumulativeAckTest.java @@ -38,7 +38,7 @@ import java.net.InetSocketAddress; import org.apache.bookkeeper.mledger.ManagedLedger; import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl; -import org.apache.pulsar.broker.TestPulsarService; +import org.apache.pulsar.broker.testcontext.PulsarTestContext; import org.apache.pulsar.broker.service.persistent.PersistentSubscription; import org.apache.pulsar.broker.service.persistent.PersistentTopic; import org.apache.pulsar.client.api.MessageId; @@ -57,14 +57,14 @@ public class MessageCumulativeAckTest { private ServerCnx serverCnx; private PersistentSubscription sub; - private TestPulsarService.Factory testPulsarServiceFactory; + private PulsarTestContext pulsarTestContext; @BeforeMethod public void setup() throws Exception { - testPulsarServiceFactory = TestPulsarService.Factory.builder() + pulsarTestContext = PulsarTestContext.builder() .build(); - serverCnx = testPulsarServiceFactory.createServerCnxSpy(); + serverCnx = pulsarTestContext.createServerCnxSpy(); doReturn(true).when(serverCnx).isActive(); doReturn(true).when(serverCnx).isWritable(); doReturn(new InetSocketAddress("localhost", 1234)).when(serverCnx).clientAddress(); @@ -74,7 +74,7 @@ public void setup() throws Exception { .when(serverCnx).getCommandSender(); String topicName = TopicName.get("MessageCumulativeAckTest").toString(); - PersistentTopic persistentTopic = new PersistentTopic(topicName, mock(ManagedLedger.class), testPulsarServiceFactory.getBrokerService()); + PersistentTopic persistentTopic = new PersistentTopic(topicName, mock(ManagedLedger.class), pulsarTestContext.getBrokerService()); sub = spy(new PersistentSubscription(persistentTopic, "sub-1", mock(ManagedCursorImpl.class), false)); doNothing().when(sub).acknowledgeMessage(any(), any(), any()); @@ -82,9 +82,9 @@ public void setup() throws Exception { @AfterMethod(alwaysRun = true) public void shutdown() throws Exception { - if (testPulsarServiceFactory != null) { - testPulsarServiceFactory.close(); - testPulsarServiceFactory = null; + if (pulsarTestContext != null) { + pulsarTestContext.close(); + pulsarTestContext = null; } sub = null; } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java index 4dfaade33cae3..d9a9dbcbcf234 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java @@ -60,7 +60,7 @@ import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl; import org.apache.bookkeeper.mledger.impl.PositionImpl; import org.apache.pulsar.broker.ServiceConfiguration; -import org.apache.pulsar.broker.TestPulsarService; +import org.apache.pulsar.broker.testcontext.PulsarTestContext; import org.apache.pulsar.broker.namespace.NamespaceService; import org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers; import org.apache.pulsar.broker.service.persistent.PersistentDispatcherSingleActiveConsumer; @@ -91,7 +91,7 @@ public class PersistentDispatcherFailoverConsumerTest { private ChannelHandlerContext channelCtx; private LinkedBlockingQueue consumerChanges; - protected TestPulsarService.Factory testPulsarServiceFactory; + protected PulsarTestContext pulsarTestContext; final String successTopicName = "persistent://part-perf/global/perf.t1/ptopic"; final String failTopicName = "persistent://part-perf/global/perf.t1/pfailTopic"; @@ -104,9 +104,9 @@ public void setup() throws Exception { svcConfig.setClusterName("pulsar-cluster"); svcConfig.setSystemTopicEnabled(false); svcConfig.setTopicLevelPoliciesEnabled(false); - testPulsarServiceFactory = TestPulsarService.Factory.builder() + pulsarTestContext = PulsarTestContext.builder() .config(svcConfig) - .useSpies(true) + .spyByDefault() .build(); consumerChanges = new LinkedBlockingQueue<>(); @@ -132,7 +132,7 @@ public void setup() throws Exception { return null; }).when(channelCtx).writeAndFlush(any(), any()); - serverCnx = testPulsarServiceFactory.createServerCnxSpy(); + serverCnx = pulsarTestContext.createServerCnxSpy(); doReturn(true).when(serverCnx).isActive(); doReturn(true).when(serverCnx).isWritable(); doReturn(new InetSocketAddress("localhost", 1234)).when(serverCnx).clientAddress(); @@ -141,7 +141,7 @@ public void setup() throws Exception { doReturn(new PulsarCommandSenderImpl(null, serverCnx)) .when(serverCnx).getCommandSender(); - serverCnxWithOldVersion = testPulsarServiceFactory.createServerCnxSpy(); + serverCnxWithOldVersion = pulsarTestContext.createServerCnxSpy(); doReturn(true).when(serverCnxWithOldVersion).isActive(); doReturn(true).when(serverCnxWithOldVersion).isWritable(); doReturn(new InetSocketAddress("localhost", 1234)) @@ -153,7 +153,7 @@ public void setup() throws Exception { .when(serverCnxWithOldVersion).getCommandSender(); NamespaceService nsSvc = mock(NamespaceService.class); - doReturn(nsSvc).when(testPulsarServiceFactory.getPulsarService()).getNamespaceService(); + doReturn(nsSvc).when(pulsarTestContext.getPulsarService()).getNamespaceService(); doReturn(true).when(nsSvc).isServiceUnitOwned(any(NamespaceBundle.class)); doReturn(true).when(nsSvc).isServiceUnitActive(any(TopicName.class)); doReturn(CompletableFuture.completedFuture(true)).when(nsSvc).checkTopicOwnership(any(TopicName.class)); @@ -164,9 +164,9 @@ public void setup() throws Exception { @AfterMethod(alwaysRun = true) public void shutdown() throws Exception { - if (testPulsarServiceFactory != null) { - testPulsarServiceFactory.close(); - testPulsarServiceFactory = null; + if (pulsarTestContext != null) { + pulsarTestContext.close(); + pulsarTestContext = null; } } @@ -181,7 +181,7 @@ void setupMLAsyncCallbackMocks() { doAnswer(invocationOnMock -> { ((OpenLedgerCallback) invocationOnMock.getArguments()[2]).openLedgerComplete(ledgerMock, null); return null; - }).when(testPulsarServiceFactory.getManagedLedgerFactory()) + }).when(pulsarTestContext.getManagedLedgerFactory()) .asyncOpen(matches(".*success.*"), any(ManagedLedgerConfig.class), any(OpenLedgerCallback.class), any(Supplier.class), any()); @@ -190,7 +190,7 @@ void setupMLAsyncCallbackMocks() { ((OpenLedgerCallback) invocationOnMock.getArguments()[2]) .openLedgerFailed(new ManagedLedgerException("Managed ledger failure"), null); return null; - }).when(testPulsarServiceFactory.getManagedLedgerFactory()) + }).when(pulsarTestContext.getManagedLedgerFactory()) .asyncOpen(matches(".*fail.*"), any(ManagedLedgerConfig.class), any(OpenLedgerCallback.class), any(Supplier.class), any()); @@ -231,7 +231,7 @@ private void verifyActiveConsumerChange(CommandActiveConsumerChange change, @Test public void testConsumerGroupChangesWithOldNewConsumers() throws Exception { PersistentTopic topic = - new PersistentTopic(successTopicName, ledgerMock, testPulsarServiceFactory.getBrokerService()); + new PersistentTopic(successTopicName, ledgerMock, pulsarTestContext.getBrokerService()); PersistentSubscription sub = new PersistentSubscription(topic, "sub-1", cursorMock, false); int partitionIndex = 0; @@ -272,7 +272,7 @@ public void testConsumerGroupChangesWithOldNewConsumers() throws Exception { public void testAddRemoveConsumer() throws Exception { log.info("--- Starting PersistentDispatcherFailoverConsumerTest::testAddConsumer ---"); - PersistentTopic topic = new PersistentTopic(successTopicName, ledgerMock, testPulsarServiceFactory.getBrokerService()); + PersistentTopic topic = new PersistentTopic(successTopicName, ledgerMock, pulsarTestContext.getBrokerService()); PersistentSubscription sub = new PersistentSubscription(topic, "sub-1", cursorMock, false); int partitionIndex = 4; @@ -404,7 +404,7 @@ public void testAddRemoveConsumerNonPartitionedTopic() throws Exception { log.info("--- Starting PersistentDispatcherFailoverConsumerTest::testAddConsumer ---"); PersistentTopic topic = - new PersistentTopic(successTopicName, ledgerMock, testPulsarServiceFactory.getBrokerService()); + new PersistentTopic(successTopicName, ledgerMock, pulsarTestContext.getBrokerService()); PersistentSubscription sub = new PersistentSubscription(topic, "sub-1", cursorMock, false); // Non partitioned topic. @@ -465,7 +465,7 @@ public void testAddRemoveConsumerNonPartitionedTopic() throws Exception { public void testMultipleDispatcherGetNextConsumerWithDifferentPriorityLevel() throws Exception { PersistentTopic topic = - new PersistentTopic(successTopicName, ledgerMock, testPulsarServiceFactory.getBrokerService()); + new PersistentTopic(successTopicName, ledgerMock, pulsarTestContext.getBrokerService()); PersistentDispatcherMultipleConsumers dispatcher = new PersistentDispatcherMultipleConsumers(topic, cursorMock, null); Consumer consumer1 = createConsumer(topic, 0, 2, false, 1); Consumer consumer2 = createConsumer(topic, 0, 2, false, 2); @@ -510,7 +510,7 @@ public void testMultipleDispatcherGetNextConsumerWithDifferentPriorityLevel() th @Test public void testFewBlockedConsumerSamePriority() throws Exception{ PersistentTopic topic = - new PersistentTopic(successTopicName, ledgerMock, testPulsarServiceFactory.getBrokerService()); + new PersistentTopic(successTopicName, ledgerMock, pulsarTestContext.getBrokerService()); PersistentDispatcherMultipleConsumers dispatcher = new PersistentDispatcherMultipleConsumers(topic, cursorMock, null); Consumer consumer1 = createConsumer(topic, 0, 2, false, 1); Consumer consumer2 = createConsumer(topic, 0, 2, false, 2); @@ -538,7 +538,7 @@ public void testFewBlockedConsumerSamePriority() throws Exception{ @Test public void testFewBlockedConsumerDifferentPriority() throws Exception { PersistentTopic topic = - new PersistentTopic(successTopicName, ledgerMock, testPulsarServiceFactory.getBrokerService()); + new PersistentTopic(successTopicName, ledgerMock, pulsarTestContext.getBrokerService()); PersistentDispatcherMultipleConsumers dispatcher = new PersistentDispatcherMultipleConsumers(topic, cursorMock, null); Consumer consumer1 = createConsumer(topic, 0, 2, false, 1); Consumer consumer2 = createConsumer(topic, 0, 2, false, 2); @@ -593,7 +593,7 @@ public void testFewBlockedConsumerDifferentPriority() throws Exception { @Test public void testFewBlockedConsumerDifferentPriority2() throws Exception { PersistentTopic topic = - new PersistentTopic(successTopicName, ledgerMock, testPulsarServiceFactory.getBrokerService()); + new PersistentTopic(successTopicName, ledgerMock, pulsarTestContext.getBrokerService()); PersistentDispatcherMultipleConsumers dispatcher = new PersistentDispatcherMultipleConsumers(topic, cursorMock, null); Consumer consumer1 = createConsumer(topic, 0, 2, true, 1); Consumer consumer2 = createConsumer(topic, 0, 2, true, 2); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java index ae9fb9a1bf71f..0bf606796638c 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java @@ -16,6 +16,7 @@ * specific language governing permissions and limitations * under the License. */ + package org.apache.pulsar.broker.service; import static org.apache.pulsar.broker.BrokerTestUtil.spyWithClassAndConstructorArgsRecordingInvocations; @@ -91,7 +92,6 @@ import org.apache.bookkeeper.test.MockedBookKeeperTestCase; import org.apache.pulsar.broker.PulsarService; import org.apache.pulsar.broker.ServiceConfiguration; -import org.apache.pulsar.broker.TestPulsarService; import org.apache.pulsar.broker.namespace.NamespaceService; import org.apache.pulsar.broker.service.persistent.CompactorSubscription; import org.apache.pulsar.broker.service.persistent.GeoPersistentReplicator; @@ -100,10 +100,12 @@ import org.apache.pulsar.broker.service.persistent.PersistentReplicator; import org.apache.pulsar.broker.service.persistent.PersistentSubscription; import org.apache.pulsar.broker.service.persistent.PersistentTopic; +import org.apache.pulsar.broker.testcontext.PulsarTestContext; 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.client.api.transaction.TxnID; +import org.apache.pulsar.client.impl.ProducerBuilderImpl; import org.apache.pulsar.client.impl.PulsarClientImpl; import org.apache.pulsar.client.impl.conf.ProducerConfigurationData; import org.apache.pulsar.common.allocator.PulsarByteBufAllocator; @@ -155,7 +157,7 @@ public class PersistentTopicTest extends MockedBookKeeperTestCase { final String successSubName2 = "successSub2"; private static final Logger log = LoggerFactory.getLogger(PersistentTopicTest.class); - protected TestPulsarService.Factory testPulsarServiceFactory; + protected PulsarTestContext pulsarTestContext; private BrokerService brokerService; @@ -170,24 +172,24 @@ public void setup() throws Exception { svcConfig.setClusterName("pulsar-cluster"); svcConfig.setTopicLevelPoliciesEnabled(false); svcConfig.setSystemTopicEnabled(false); - testPulsarServiceFactory = TestPulsarService.Factory.builder() + pulsarTestContext = PulsarTestContext.builder() .config(svcConfig) - .useSpies(true) + .spyByDefault() .useTestPulsarResources(metadataStore) .compactor(mock(Compactor.class)) .build(); - brokerService = testPulsarServiceFactory.getBrokerService(); + brokerService = pulsarTestContext.getBrokerService(); doAnswer(invocationOnMock -> CompletableFuture.completedFuture(null)) - .when(testPulsarServiceFactory.getManagedLedgerFactory()).getManagedLedgerPropertiesAsync(any()); + .when(pulsarTestContext.getManagedLedgerFactory()).getManagedLedgerPropertiesAsync(any()); doAnswer(invocation -> { DeleteLedgerCallback deleteLedgerCallback = invocation.getArgument(1); deleteLedgerCallback.deleteLedgerComplete(null); return null; - }).when(testPulsarServiceFactory.getManagedLedgerFactory()).asyncDelete(any(), any(), any()); + }).when(pulsarTestContext.getManagedLedgerFactory()).asyncDelete(any(), any(), any()); // Mock serviceCnx. serverCnx = spyWithClassAndConstructorArgsRecordingInvocations(ServerCnx.class, - testPulsarServiceFactory.getPulsarService()); + pulsarTestContext.getPulsarService()); doReturn(true).when(serverCnx).isActive(); doReturn(true).when(serverCnx).isWritable(); doReturn(new InetSocketAddress("localhost", 1234)).when(serverCnx).clientAddress(); @@ -202,7 +204,7 @@ public void setup() throws Exception { NamespaceService nsSvc = mock(NamespaceService.class); NamespaceBundle bundle = mock(NamespaceBundle.class); doReturn(CompletableFuture.completedFuture(bundle)).when(nsSvc).getBundleAsync(any()); - doReturn(nsSvc).when(testPulsarServiceFactory.getPulsarService()).getNamespaceService(); + doReturn(nsSvc).when(pulsarTestContext.getPulsarService()).getNamespaceService(); doReturn(true).when(nsSvc).isServiceUnitOwned(any()); doReturn(true).when(nsSvc).isServiceUnitActive(any()); doReturn(CompletableFuture.completedFuture(true)).when(nsSvc).isServiceUnitActiveAsync(any()); @@ -213,9 +215,9 @@ public void setup() throws Exception { @AfterMethod(alwaysRun = true) public void teardown() throws Exception { - if (testPulsarServiceFactory != null) { - testPulsarServiceFactory.close(); - testPulsarServiceFactory = null; + if (pulsarTestContext != null) { + pulsarTestContext.close(); + pulsarTestContext = null; } } @@ -228,7 +230,7 @@ public void testCreateTopic() { doAnswer(invocationOnMock -> { ((OpenLedgerCallback) invocationOnMock.getArguments()[2]).openLedgerComplete(ledgerMock, null); return null; - }).when(testPulsarServiceFactory.getManagedLedgerFactory()) + }).when(pulsarTestContext.getManagedLedgerFactory()) .asyncOpen(anyString(), any(ManagedLedgerConfig.class), any(OpenLedgerCallback.class), any(Supplier.class), any()); @@ -254,7 +256,7 @@ public void testCreateTopicMLFailure() { .openLedgerFailed(new ManagedLedgerException("Managed ledger failure"), null)).start(); return null; - }).when(testPulsarServiceFactory.getManagedLedgerFactory()) + }).when(pulsarTestContext.getManagedLedgerFactory()) .asyncOpen(anyString(), any(ManagedLedgerConfig.class), any(OpenLedgerCallback.class), any(Supplier.class), any()); @@ -315,19 +317,24 @@ public void setMetadataFromEntryData(ByteBuf entryData) { @Test public void testDispatcherMultiConsumerReadFailed() { - PersistentTopic topic = spyWithClassAndConstructorArgsRecordingInvocations(PersistentTopic.class, successTopicName, ledgerMock, brokerService); + PersistentTopic topic = + spyWithClassAndConstructorArgsRecordingInvocations(PersistentTopic.class, successTopicName, ledgerMock, + brokerService); ManagedCursor cursor = mock(ManagedCursor.class); when(cursor.getName()).thenReturn("cursor"); Subscription subscription = mock(Subscription.class); when(subscription.getName()).thenReturn("sub"); - PersistentDispatcherMultipleConsumers dispatcher = new PersistentDispatcherMultipleConsumers(topic, cursor, subscription); + PersistentDispatcherMultipleConsumers dispatcher = + new PersistentDispatcherMultipleConsumers(topic, cursor, subscription); dispatcher.readEntriesFailed(new ManagedLedgerException.InvalidCursorPositionException("failed"), null); verify(topic, atLeast(1)).getBrokerService(); } @Test public void testDispatcherSingleConsumerReadFailed() { - PersistentTopic topic = spyWithClassAndConstructorArgsRecordingInvocations(PersistentTopic.class, successTopicName, ledgerMock, brokerService); + PersistentTopic topic = + spyWithClassAndConstructorArgsRecordingInvocations(PersistentTopic.class, successTopicName, ledgerMock, + brokerService); ManagedCursor cursor = mock(ManagedCursor.class); when(cursor.getName()).thenReturn("cursor"); PersistentDispatcherSingleActiveConsumer dispatcher = new PersistentDispatcherSingleActiveConsumer(cursor, @@ -517,7 +524,7 @@ private void testMaxProducers() { @Test public void testMaxProducersForBroker() { // set max clients - testPulsarServiceFactory.getConfig().setMaxProducersPerTopic(2); + pulsarTestContext.getConfig().setMaxProducersPerTopic(2); testMaxProducers(); } @@ -526,9 +533,9 @@ public void testMaxProducersForNamespace() throws Exception { // set max clients Policies policies = new Policies(); policies.max_producers_per_topic = 2; - testPulsarServiceFactory.getPulsarResources().getNamespaceResources() - .createPolicies(TopicName.get(successTopicName).getNamespaceObject(), - policies); + pulsarTestContext.getPulsarResources().getNamespaceResources() + .createPolicies(TopicName.get(successTopicName).getNamespaceObject(), + policies); testMaxProducers(); } @@ -536,7 +543,7 @@ private Producer getMockedProducerWithSpecificAddress(Topic topic, long producer final String producerNameBase = "producer"; final String role = "appid1"; - ServerCnx cnx = testPulsarServiceFactory.createServerCnxSpy(); + ServerCnx cnx = pulsarTestContext.createServerCnxSpy(); doReturn(true).when(cnx).isActive(); doReturn(true).when(cnx).isWritable(); doReturn(new InetSocketAddress(address, 1234)).when(cnx).clientAddress(); @@ -550,7 +557,7 @@ private Producer getMockedProducerWithSpecificAddress(Topic topic, long producer @Test public void testMaxSameAddressProducers() throws Exception { // set max clients - testPulsarServiceFactory.getConfig().setMaxSameAddressProducersPerTopic(2); + pulsarTestContext.getConfig().setMaxSameAddressProducersPerTopic(2); PersistentTopic topic = new PersistentTopic(successTopicName, ledgerMock, brokerService); @@ -694,7 +701,8 @@ public void testChangeSubscriptionType() throws Exception { Consumer consumer = new Consumer(sub, SubType.Exclusive, topic.getName(), 1, 0, "Cons1", true, serverCnx, "myrole-1", Collections.emptyMap(), false, - new KeySharedMeta().setKeySharedMode(KeySharedMode.AUTO_SPLIT), MessageId.latest, DEFAULT_CONSUMER_EPOCH); + new KeySharedMeta().setKeySharedMode(KeySharedMode.AUTO_SPLIT), MessageId.latest, + DEFAULT_CONSUMER_EPOCH); sub.addConsumer(consumer); consumer.close(); @@ -705,7 +713,8 @@ public void testChangeSubscriptionType() throws Exception { consumer = new Consumer(sub, subType, topic.getName(), 1, 0, "Cons1", true, serverCnx, "myrole-1", Collections.emptyMap(), false, - new KeySharedMeta().setKeySharedMode(KeySharedMode.AUTO_SPLIT), MessageId.latest, DEFAULT_CONSUMER_EPOCH); + new KeySharedMeta().setKeySharedMode(KeySharedMode.AUTO_SPLIT), MessageId.latest, + DEFAULT_CONSUMER_EPOCH); sub.addConsumer(consumer); assertTrue(sub.getDispatcher().isConsumerConnected()); @@ -727,8 +736,10 @@ public void testAddRemoveConsumer() throws Exception { PersistentSubscription sub = new PersistentSubscription(topic, "sub-1", cursorMock, false); // 1. simple add consumer - Consumer consumer = new Consumer(sub, SubType.Exclusive, topic.getName(), 1 /* consumer id */, 0, "Cons1"/* consumer name */, - true, serverCnx, "myrole-1", Collections.emptyMap(), false /* read compacted */, null, MessageId.latest, DEFAULT_CONSUMER_EPOCH); + Consumer consumer = new Consumer(sub, SubType.Exclusive, topic.getName(), 1 /* consumer id */, 0, + "Cons1"/* consumer name */, + true, serverCnx, "myrole-1", Collections.emptyMap(), false /* read compacted */, null, MessageId.latest, + DEFAULT_CONSUMER_EPOCH); sub.addConsumer(consumer); assertTrue(sub.getDispatcher().isConsumerConnected()); @@ -853,8 +864,8 @@ private void testMaxConsumersShared() throws Exception { @Test public void testMaxConsumersSharedForBroker() throws Exception { // set max clients - testPulsarServiceFactory.getConfig().setMaxConsumersPerSubscription(2); - testPulsarServiceFactory.getConfig().setMaxConsumersPerTopic(3); + pulsarTestContext.getConfig().setMaxConsumersPerSubscription(2); + pulsarTestContext.getConfig().setMaxConsumersPerTopic(3); testMaxConsumersShared(); } @@ -865,7 +876,7 @@ public void testMaxConsumersSharedForNamespace() throws Exception { policies.max_consumers_per_subscription = 2; policies.max_consumers_per_topic = 3; - testPulsarServiceFactory.getPulsarResources().getNamespaceResources() + pulsarTestContext.getPulsarResources().getNamespaceResources() .createPolicies(TopicName.get(successTopicName).getNamespaceObject(), policies); @@ -949,8 +960,8 @@ private void testMaxConsumersFailover() throws Exception { @Test public void testMaxConsumersFailoverForBroker() throws Exception { // set max clients - testPulsarServiceFactory.getConfig().setMaxConsumersPerSubscription(2); - testPulsarServiceFactory.getConfig().setMaxConsumersPerTopic(3); + pulsarTestContext.getConfig().setMaxConsumersPerSubscription(2); + pulsarTestContext.getConfig().setMaxConsumersPerTopic(3); testMaxConsumersFailover(); } @@ -962,7 +973,7 @@ public void testMaxConsumersFailoverForNamespace() throws Exception { policies.max_consumers_per_subscription = 2; policies.max_consumers_per_topic = 3; - testPulsarServiceFactory.getPulsarResources().getNamespaceResources() + pulsarTestContext.getPulsarResources().getNamespaceResources() .createPolicies(TopicName.get(successTopicName).getNamespaceObject(), policies); @@ -974,7 +985,7 @@ private Consumer getMockedConsumerWithSpecificAddress(Topic topic, Subscription final String consumerNameBase = "consumer"; final String role = "appid1"; - ServerCnx cnx = testPulsarServiceFactory.createServerCnxSpy(); + ServerCnx cnx = pulsarTestContext.createServerCnxSpy(); doReturn(true).when(cnx).isActive(); doReturn(true).when(cnx).isWritable(); doReturn(new InetSocketAddress(address, 1234)).when(cnx).clientAddress(); @@ -988,7 +999,7 @@ private Consumer getMockedConsumerWithSpecificAddress(Topic topic, Subscription @Test public void testMaxSameAddressConsumers() throws Exception { // set max clients - testPulsarServiceFactory.getConfig().setMaxSameAddressConsumersPerTopic(2); + pulsarTestContext.getConfig().setMaxSameAddressConsumersPerTopic(2); PersistentTopic topic = new PersistentTopic(successTopicName, ledgerMock, brokerService); PersistentSubscription sub1 = new PersistentSubscription(topic, "sub1", cursorMock, false); @@ -1091,8 +1102,10 @@ public void testMaxSameAddressConsumers() throws Exception { public void testUbsubscribeRaceConditions() throws Exception { PersistentTopic topic = new PersistentTopic(successTopicName, ledgerMock, brokerService); PersistentSubscription sub = new PersistentSubscription(topic, "sub-1", cursorMock, false); - Consumer consumer1 = new Consumer(sub, SubType.Exclusive, topic.getName(), 1 /* consumer id */, 0, "Cons1"/* consumer name */, - true, serverCnx, "myrole-1", Collections.emptyMap(), false /* read compacted */, null, MessageId.latest, DEFAULT_CONSUMER_EPOCH); + Consumer consumer1 = new Consumer(sub, SubType.Exclusive, topic.getName(), 1 /* consumer id */, 0, + "Cons1"/* consumer name */, + true, serverCnx, "myrole-1", Collections.emptyMap(), false /* read compacted */, null, MessageId.latest, + DEFAULT_CONSUMER_EPOCH); sub.addConsumer(consumer1); doAnswer(invocationOnMock -> { @@ -1113,7 +1126,8 @@ public void testUbsubscribeRaceConditions() throws Exception { Thread.sleep(10); /* delay to ensure that the ubsubscribe gets executed first */ sub.addConsumer(new Consumer(sub, SubType.Exclusive, topic.getName(), 2 /* consumer id */, 0, "Cons2"/* consumer name */, true, serverCnx, - "myrole-1", Collections.emptyMap(), false /* read compacted */, null, MessageId.latest, DEFAULT_CONSUMER_EPOCH)).get(); + "myrole-1", Collections.emptyMap(), false /* read compacted */, null, MessageId.latest, + DEFAULT_CONSUMER_EPOCH)).get(); fail(); } catch (Exception e) { assertTrue(e.getCause() instanceof BrokerServiceException.SubscriptionFencedException); @@ -1350,7 +1364,8 @@ public void testDeleteTopicRaceConditions() throws Exception { String role = "appid1"; Thread.sleep(10); /* delay to ensure that the delete gets executed first */ Producer producer = new Producer(topic, serverCnx, 1 /* producer id */, "prod-name", - role, false, null, SchemaVersion.Latest, 0, false, ProducerAccessMode.Shared, Optional.empty(), true); + role, false, null, SchemaVersion.Latest, 0, false, ProducerAccessMode.Shared, Optional.empty(), + true); topic.addProducer(producer, new CompletableFuture<>()).join(); fail("Should have failed"); } catch (Exception e) { @@ -1412,7 +1427,7 @@ public void closeFailed(ManagedLedgerException exception, Object ctx) { doAnswer(invocationOnMock -> { ((OpenLedgerCallback) invocationOnMock.getArguments()[2]).openLedgerComplete(ledgerMock, null); return null; - }).when(testPulsarServiceFactory.getManagedLedgerFactory()) + }).when(pulsarTestContext.getManagedLedgerFactory()) .asyncOpen(matches(".*success.*"), any(ManagedLedgerConfig.class), any(OpenLedgerCallback.class), any(Supplier.class), any()); @@ -1421,7 +1436,7 @@ public void closeFailed(ManagedLedgerException exception, Object ctx) { ((OpenLedgerCallback) invocationOnMock.getArguments()[2]) .openLedgerFailed(new ManagedLedgerException("Managed ledger failure"), null); return null; - }).when(testPulsarServiceFactory.getManagedLedgerFactory()) + }).when(pulsarTestContext.getManagedLedgerFactory()) .asyncOpen(matches(".*fail.*"), any(ManagedLedgerConfig.class), any(OpenLedgerCallback.class), any(Supplier.class), any()); @@ -1437,7 +1452,9 @@ public void closeFailed(ManagedLedgerException exception, Object ctx) { doAnswer(invocationOnMock -> { ((OpenCursorCallback) invocationOnMock.getArguments()[2]).openCursorComplete(cursorMock, null); return null; - }).when(ledgerMock).asyncOpenCursor(matches(".*success.*"), any(InitialPosition.class), any(OpenCursorCallback.class), any()); + }).when(ledgerMock) + .asyncOpenCursor(matches(".*success.*"), any(InitialPosition.class), any(OpenCursorCallback.class), + any()); doAnswer(invocationOnMock -> { ((OpenCursorCallback) invocationOnMock.getArguments()[4]).openCursorComplete(cursorMock, null); @@ -1462,10 +1479,10 @@ public void closeFailed(ManagedLedgerException exception, Object ctx) { }).when(ledgerMock).asyncDeleteCursor(matches(".*success.*"), any(DeleteCursorCallback.class), any()); doAnswer((invokactionOnMock) -> { - ((MarkDeleteCallback) invokactionOnMock.getArguments()[2]) + ((MarkDeleteCallback) invokactionOnMock.getArguments()[2]) .markDeleteComplete(invokactionOnMock.getArguments()[3]); - return null; - }).when(cursorMock).asyncMarkDelete(any(), any(), any(MarkDeleteCallback.class), any()); + return null; + }).when(cursorMock).asyncMarkDelete(any(), any(), any(MarkDeleteCallback.class), any()); } @@ -1683,20 +1700,26 @@ public void testAtomicReplicationRemoval() throws Exception { String remoteReplicatorName = topic.getReplicatorPrefix() + "." + remoteCluster; ConcurrentOpenHashMap replicatorMap = topic.getReplicators(); - PulsarService pulsar = testPulsarServiceFactory.getPulsarService(); - final URL brokerUrl = new URL( - "http://" + pulsar.getAdvertisedAddress() + ":" + pulsar.getConfiguration().getBrokerServicePort() - .get()); - @Cleanup - PulsarClient client = PulsarClient.builder().serviceUrl(brokerUrl.toString()).build(); ManagedCursor cursor = mock(ManagedCursorImpl.class); doReturn(remoteCluster).when(cursor).getName(); - brokerService.getReplicationClients().put(remoteCluster, client); + PulsarClientImpl pulsarClientMock = mock(PulsarClientImpl.class); + when(pulsarClientMock.newProducer(any())).thenAnswer( + invocation -> { + ProducerBuilderImpl producerBuilder = + new ProducerBuilderImpl(pulsarClientMock, invocation.getArgument(0)) { + @Override + public CompletableFuture createAsync() { + return CompletableFuture.completedFuture(mock(org.apache.pulsar.client.api.Producer.class)); + } + }; + return producerBuilder; + }); + brokerService.getReplicationClients().put(remoteCluster, pulsarClientMock); PersistentReplicator replicator = spy( new GeoPersistentReplicator(topic, cursor, localCluster, remoteCluster, brokerService, (PulsarClientImpl) brokerService.getReplicationClient(remoteCluster, brokerService.pulsar().getPulsarResources().getClusterResources() - .getCluster(remoteCluster)))); + .getCluster(remoteCluster)))); replicatorMap.put(remoteReplicatorName, replicator); // step-1 remove replicator : it will disconnect the producer but it will wait for callback to be completed @@ -1730,7 +1753,7 @@ public void testClosingReplicationProducerTwice() throws Exception { PersistentTopic topic = new PersistentTopic(globalTopicName, ledgerMock, brokerService); - PulsarService pulsar = testPulsarServiceFactory.getPulsarService(); + PulsarService pulsar = pulsarTestContext.getPulsarService(); final URL brokerUrl = new URL( "http://" + pulsar.getAdvertisedAddress() + ":" + pulsar.getConfiguration().getBrokerServicePort() .get()); @@ -1738,7 +1761,7 @@ public void testClosingReplicationProducerTwice() throws Exception { PulsarClient client = spy(PulsarClient.builder().serviceUrl(brokerUrl.toString()).build()); PulsarClientImpl clientImpl = (PulsarClientImpl) client; doReturn(new CompletableFuture()).when(clientImpl) - .createProducerAsync(any(ProducerConfigurationData.class), any(Schema.class)); + .createProducerAsync(any(ProducerConfigurationData.class), any(Schema.class)); ManagedCursor cursor = mock(ManagedCursorImpl.class); doReturn(remoteCluster).when(cursor).getName(); @@ -1750,10 +1773,10 @@ public void testClosingReplicationProducerTwice() throws Exception { // PersistentReplicator constructor calls startProducer() verify(clientImpl) - .createProducerAsync( - any(ProducerConfigurationData.class), - any(), eq(null) - ); + .createProducerAsync( + any(ProducerConfigurationData.class), + any(), eq(null) + ); replicator.disconnect(false); replicator.disconnect(false); @@ -1770,8 +1793,8 @@ public void testCompactorSubscription() { when(compactedTopic.newCompactedLedger(any(Position.class), anyLong())) .thenReturn(CompletableFuture.completedFuture(mock(CompactedTopicContext.class))); PersistentSubscription sub = new CompactorSubscription(topic, compactedTopic, - Compactor.COMPACTION_SUBSCRIPTION, - cursorMock); + Compactor.COMPACTION_SUBSCRIPTION, + cursorMock); PositionImpl position = new PositionImpl(1, 1); long ledgerId = 0xc0bfefeL; sub.acknowledgeMessage(Collections.singletonList(position), AckType.Cumulative, @@ -1800,13 +1823,13 @@ public void testCompactorSubscriptionUpdatedOnInit() { @Test public void testCompactionTriggeredAfterThresholdFirstInvocation() throws Exception { CompletableFuture compactPromise = new CompletableFuture<>(); - Compactor compactor = testPulsarServiceFactory.getPulsarService().getCompactor(); + Compactor compactor = pulsarTestContext.getPulsarService().getCompactor(); doReturn(compactPromise).when(compactor).compact(anyString()); Policies policies = new Policies(); policies.compaction_threshold = 1L; - testPulsarServiceFactory.getPulsarResources().getNamespaceResources() + pulsarTestContext.getPulsarResources().getNamespaceResources() .createPolicies(TopicName.get(successTopicName).getNamespaceObject(), policies); @@ -1831,7 +1854,7 @@ public void testCompactionTriggeredAfterThresholdFirstInvocation() throws Except @Test public void testCompactionTriggeredAfterThresholdSecondInvocation() throws Exception { CompletableFuture compactPromise = new CompletableFuture<>(); - Compactor compactor = testPulsarServiceFactory.getPulsarService().getCompactor(); + Compactor compactor = pulsarTestContext.getPulsarService().getCompactor(); doReturn(compactPromise).when(compactor).compact(anyString()); ManagedCursor subCursor = mock(ManagedCursor.class); @@ -1841,7 +1864,7 @@ public void testCompactionTriggeredAfterThresholdSecondInvocation() throws Excep Policies policies = new Policies(); policies.compaction_threshold = 1L; - testPulsarServiceFactory.getPulsarResources().getNamespaceResources() + pulsarTestContext.getPulsarResources().getNamespaceResources() .createPolicies(TopicName.get(successTopicName).getNamespaceObject(), policies); @@ -1865,13 +1888,13 @@ public void testCompactionTriggeredAfterThresholdSecondInvocation() throws Excep @Test public void testCompactionDisabledWithZeroThreshold() throws Exception { CompletableFuture compactPromise = new CompletableFuture<>(); - Compactor compactor = testPulsarServiceFactory.getPulsarService().getCompactor(); + Compactor compactor = pulsarTestContext.getPulsarService().getCompactor(); doReturn(compactPromise).when(compactor).compact(anyString()); Policies policies = new Policies(); policies.compaction_threshold = 0L; - testPulsarServiceFactory.getPulsarResources().getNamespaceResources() + pulsarTestContext.getPulsarResources().getNamespaceResources() .createPolicies(TopicName.get(successTopicName).getNamespaceObject(), policies); @@ -1886,7 +1909,7 @@ public void testCompactionDisabledWithZeroThreshold() throws Exception { @Test public void testBacklogCursor() throws Exception { int backloggedThreshold = 10; - testPulsarServiceFactory.getConfig().setManagedLedgerCursorBackloggedThreshold(backloggedThreshold); + pulsarTestContext.getConfig().setManagedLedgerCursorBackloggedThreshold(backloggedThreshold); ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open("cache_backlog_ledger"); PersistentTopic topic = new PersistentTopic(successTopicName, ledger, brokerService); @@ -1895,22 +1918,28 @@ public void testBacklogCursor() throws Exception { // Open cursor1, add it into activeCursor-container and add it into subscription consumer list ManagedCursor cursor1 = ledger.openCursor("c1"); PersistentSubscription sub1 = new PersistentSubscription(topic, "sub-1", cursor1, false); - Consumer consumer1 = new Consumer(sub1, SubType.Exclusive, topic.getName(), 1 /* consumer id */, 0, "Cons1"/* consumer name */, - true, serverCnx, "myrole-1", Collections.emptyMap(), false /* read compacted */, null, MessageId.latest, DEFAULT_CONSUMER_EPOCH); + Consumer consumer1 = new Consumer(sub1, SubType.Exclusive, topic.getName(), 1 /* consumer id */, 0, + "Cons1"/* consumer name */, + true, serverCnx, "myrole-1", Collections.emptyMap(), false /* read compacted */, null, MessageId.latest, + DEFAULT_CONSUMER_EPOCH); topic.getSubscriptions().put(Codec.decode(cursor1.getName()), sub1); sub1.addConsumer(consumer1); // Open cursor2, add it into activeCursor-container and add it into subscription consumer list ManagedCursor cursor2 = ledger.openCursor("c2"); PersistentSubscription sub2 = new PersistentSubscription(topic, "sub-2", cursor2, false); - Consumer consumer2 = new Consumer(sub2, SubType.Exclusive, topic.getName(), 2 /* consumer id */, 0, "Cons2"/* consumer name */, - true, serverCnx, "myrole-2", Collections.emptyMap(), false /* read compacted */, null, MessageId.latest, DEFAULT_CONSUMER_EPOCH); + Consumer consumer2 = new Consumer(sub2, SubType.Exclusive, topic.getName(), 2 /* consumer id */, 0, + "Cons2"/* consumer name */, + true, serverCnx, "myrole-2", Collections.emptyMap(), false /* read compacted */, null, MessageId.latest, + DEFAULT_CONSUMER_EPOCH); topic.getSubscriptions().put(Codec.decode(cursor2.getName()), sub2); sub2.addConsumer(consumer2); // Open cursor3, add it into activeCursor-container and do not add it into subscription consumer list ManagedCursor cursor3 = ledger.openCursor("c3"); PersistentSubscription sub3 = new PersistentSubscription(topic, "sub-3", cursor3, false); - Consumer consumer3 = new Consumer(sub2, SubType.Exclusive, topic.getName(), 3 /* consumer id */, 0, "Cons2"/* consumer name */, - true, serverCnx, "myrole-3", Collections.emptyMap(), false /* read compacted */, null, MessageId.latest, DEFAULT_CONSUMER_EPOCH); + Consumer consumer3 = new Consumer(sub2, SubType.Exclusive, topic.getName(), 3 /* consumer id */, 0, + "Cons2"/* consumer name */, + true, serverCnx, "myrole-3", Collections.emptyMap(), false /* read compacted */, null, MessageId.latest, + DEFAULT_CONSUMER_EPOCH); topic.getSubscriptions().put(Codec.decode(cursor3.getName()), sub3); // Case1: cursors are active as haven't started deactivateBacklogCursor scan @@ -2006,13 +2035,19 @@ public void testCheckInactiveSubscriptions() throws Exception { .concurrencyLevel(1) .build(); // This subscription is connected by consumer. - PersistentSubscription nonDeletableSubscription1 = spyWithClassAndConstructorArgsRecordingInvocations(PersistentSubscription.class, topic, "nonDeletableSubscription1", cursorMock, false); + PersistentSubscription nonDeletableSubscription1 = + spyWithClassAndConstructorArgsRecordingInvocations(PersistentSubscription.class, topic, + "nonDeletableSubscription1", cursorMock, false); subscriptions.put(nonDeletableSubscription1.getName(), nonDeletableSubscription1); // This subscription is not connected by consumer. - PersistentSubscription deletableSubscription1 = spyWithClassAndConstructorArgsRecordingInvocations(PersistentSubscription.class, topic, "deletableSubscription1", cursorMock, false); + PersistentSubscription deletableSubscription1 = + spyWithClassAndConstructorArgsRecordingInvocations(PersistentSubscription.class, topic, + "deletableSubscription1", cursorMock, false); subscriptions.put(deletableSubscription1.getName(), deletableSubscription1); // This subscription is replicated. - PersistentSubscription nonDeletableSubscription2 = spyWithClassAndConstructorArgsRecordingInvocations(PersistentSubscription.class, topic, "nonDeletableSubscription2", cursorMock, true); + PersistentSubscription nonDeletableSubscription2 = + spyWithClassAndConstructorArgsRecordingInvocations(PersistentSubscription.class, topic, + "nonDeletableSubscription2", cursorMock, true); subscriptions.put(nonDeletableSubscription2.getName(), nonDeletableSubscription2); Field field = topic.getClass().getDeclaredField("subscriptions"); @@ -2027,11 +2062,11 @@ public void testCheckInactiveSubscriptions() throws Exception { true, serverCnx, "app1", Collections.emptyMap(), false, null, MessageId.latest, DEFAULT_CONSUMER_EPOCH); addConsumerToSubscription.invoke(topic, nonDeletableSubscription1, consumer); - testPulsarServiceFactory.getPulsarResources().getNamespaceResources() + pulsarTestContext.getPulsarResources().getNamespaceResources() .createPolicies(TopicName.get(successTopicName).getNamespaceObject(), new Policies()); - testPulsarServiceFactory.getConfig().setSubscriptionExpirationTimeMinutes(5); + pulsarTestContext.getConfig().setSubscriptionExpirationTimeMinutes(5); doReturn(System.currentTimeMillis() - TimeUnit.MINUTES.toMillis(6)).when(cursorMock).getLastActive(); @@ -2058,7 +2093,7 @@ public void testTopicFencingTimeout() throws Exception { Field isClosingOrDeletingField = PersistentTopic.class.getDeclaredField("isClosingOrDeleting"); isClosingOrDeletingField.setAccessible(true); - testPulsarServiceFactory.getConfig().setTopicFencingTimeoutSeconds(10); + pulsarTestContext.getConfig().setTopicFencingTimeoutSeconds(10); fence.invoke(topic); unfence.invoke(topic); ScheduledFuture fencedTopicMonitoringTask = (ScheduledFuture) fencedTopicMonitoringTaskField.get(topic); @@ -2067,7 +2102,7 @@ public void testTopicFencingTimeout() throws Exception { assertFalse((boolean) isFencedField.get(topic)); assertFalse((boolean) isClosingOrDeletingField.get(topic)); - testPulsarServiceFactory.getConfig().setTopicFencingTimeoutSeconds(1); + pulsarTestContext.getConfig().setTopicFencingTimeoutSeconds(1); fence.invoke(topic); Thread.sleep(2000); fencedTopicMonitoringTask = (ScheduledFuture) fencedTopicMonitoringTaskField.get(topic); @@ -2079,7 +2114,7 @@ public void testTopicFencingTimeout() throws Exception { @Test public void testTopicCloseFencingTimeout() throws Exception { - testPulsarServiceFactory.getConfig().setTopicFencingTimeoutSeconds(10); + pulsarTestContext.getConfig().setTopicFencingTimeoutSeconds(10); Method fence = PersistentTopic.class.getDeclaredMethod("fence"); fence.setAccessible(true); Field fencedTopicMonitoringTaskField = PersistentTopic.class.getDeclaredField("fencedTopicMonitoringTask"); @@ -2106,7 +2141,8 @@ public void testGetDurableSubscription() throws Exception { Position mockPosition = mock(Position.class); doReturn("test").when(mockCursor).getName(); doAnswer((Answer) invocationOnMock -> { - ((AsyncCallbacks.FindEntryCallback) invocationOnMock.getArguments()[2]).findEntryComplete(mockPosition, invocationOnMock.getArguments()[3]); + ((AsyncCallbacks.FindEntryCallback) invocationOnMock.getArguments()[2]).findEntryComplete(mockPosition, + invocationOnMock.getArguments()[3]); return null; }).when(mockCursor).asyncFindNewestMatching(any(), any(), any(), any()); doAnswer((Answer) invocationOnMock -> { @@ -2184,7 +2220,7 @@ public void testKeySharedMetadataExposedToStats() throws Exception { assertEquals(stats2.keySharedMode, "AUTO_SPLIT"); assertTrue(stats2.allowOutOfOrderDelivery); - KeySharedMeta ksm = new KeySharedMeta().setKeySharedMode(KeySharedMode.STICKY) + KeySharedMeta ksm = new KeySharedMeta().setKeySharedMode(KeySharedMode.STICKY) .setAllowOutOfOrderDelivery(false); ksm.addHashRange().setStart(0).setEnd(65535); Consumer consumer3 = new Consumer(sub3, SubType.Key_Shared, topic.getName(), 3, 0, "Cons3", true, serverCnx, @@ -2222,7 +2258,7 @@ public void testGetReplicationClusters() throws MetadataStoreException { Set namespaceClusters = new HashSet<>(); namespaceClusters.add("namespace-cluster"); policies.replication_clusters = namespaceClusters; - testPulsarServiceFactory.getPulsarResources().getNamespaceResources() + pulsarTestContext.getPulsarResources().getNamespaceResources() .createPolicies(TopicName.get(successTopicName).getNamespaceObject(), policies); topic = new PersistentTopic(successTopicName, ledgerMock, brokerService); @@ -2230,7 +2266,7 @@ public void testGetReplicationClusters() throws MetadataStoreException { assertEquals(topic.getHierarchyTopicPolicies().getReplicationClusters().get(), namespaceClusters); TopicPoliciesService topicPoliciesService = mock(TopicPoliciesService.class); - doReturn(topicPoliciesService).when(testPulsarServiceFactory.getPulsarService()).getTopicPoliciesService(); + doReturn(topicPoliciesService).when(pulsarTestContext.getPulsarService()).getTopicPoliciesService(); CompletableFuture> topicPoliciesFuture = new CompletableFuture<>(); TopicPolicies topicPolicies = new TopicPolicies(); List topicClusters = new ArrayList<>(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxAuthorizationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxAuthorizationTest.java index 325cb8d84e43f..c1e32c0886e26 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxAuthorizationTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxAuthorizationTest.java @@ -45,7 +45,7 @@ import java.util.Properties; import javax.crypto.SecretKey; import org.apache.pulsar.broker.ServiceConfiguration; -import org.apache.pulsar.broker.TestPulsarService; +import org.apache.pulsar.broker.testcontext.PulsarTestContext; import org.apache.pulsar.broker.authentication.AuthenticationDataSource; import org.apache.pulsar.broker.authentication.AuthenticationDataSubscription; import org.apache.pulsar.broker.authentication.AuthenticationProviderToken; @@ -76,7 +76,7 @@ public class ServerCnxAuthorizationTest { private ServiceConfiguration svcConfig; - protected TestPulsarService.Factory testPulsarServiceFactory; + protected PulsarTestContext pulsarTestContext; private BrokerService brokerService; @BeforeMethod(alwaysRun = true) @@ -95,21 +95,21 @@ public void beforeMethod() throws Exception { properties.setProperty("tokenSecretKey", "data:;base64," + Base64.getEncoder().encodeToString(SECRET_KEY.getEncoded())); svcConfig.setProperties(properties); - testPulsarServiceFactory = TestPulsarService.Factory.builder() + pulsarTestContext = PulsarTestContext.builder() .config(svcConfig) - .useSpies(true) + .spyByDefault() .build(); - brokerService = testPulsarServiceFactory.getBrokerService(); + brokerService = pulsarTestContext.getBrokerService(); - testPulsarServiceFactory.getPulsarResources().getTenantResources().createTenant("public", + pulsarTestContext.getPulsarResources().getTenantResources().createTenant("public", TenantInfo.builder().build()); } @AfterMethod(alwaysRun = true) public void cleanup() throws Exception { - if (testPulsarServiceFactory != null) { - testPulsarServiceFactory.close(); - testPulsarServiceFactory = null; + if (pulsarTestContext != null) { + pulsarTestContext.close(); + pulsarTestContext = null; } } @@ -118,7 +118,7 @@ public void testVerifyOriginalPrincipalWithAuthDataForwardedFromProxy() throws E svcConfig.setAuthenticateOriginalAuthData(true); - ServerCnx serverCnx = testPulsarServiceFactory.createServerCnxSpy(); + ServerCnx serverCnx = pulsarTestContext.createServerCnxSpy(); ChannelHandlerContext channelHandlerContext = mock(ChannelHandlerContext.class); Channel channel = mock(Channel.class); ChannelPipeline channelPipeline = mock(ChannelPipeline.class); @@ -155,7 +155,7 @@ public void testVerifyOriginalPrincipalWithAuthDataForwardedFromProxy() throws E AuthorizationService authorizationService = spyWithClassAndConstructorArgsRecordingInvocations(AuthorizationService.class, svcConfig, - testPulsarServiceFactory.getPulsarResources()); + pulsarTestContext.getPulsarResources()); doReturn(authorizationService).when(brokerService).getAuthorizationService(); // lookup @@ -225,7 +225,7 @@ public void testVerifyOriginalPrincipalWithAuthDataForwardedFromProxy() throws E public void testVerifyOriginalPrincipalWithoutAuthDataForwardedFromProxy() throws Exception { svcConfig.setAuthenticateOriginalAuthData(false); - ServerCnx serverCnx = testPulsarServiceFactory.createServerCnxSpy(); + ServerCnx serverCnx = pulsarTestContext.createServerCnxSpy(); ChannelHandlerContext channelHandlerContext = mock(ChannelHandlerContext.class); Channel channel = mock(Channel.class); ChannelPipeline channelPipeline = mock(ChannelPipeline.class); @@ -257,7 +257,7 @@ public void testVerifyOriginalPrincipalWithoutAuthDataForwardedFromProxy() throw AuthorizationService authorizationService = spyWithClassAndConstructorArgsRecordingInvocations(AuthorizationService.class, svcConfig, - testPulsarServiceFactory.getPulsarResources()); + pulsarTestContext.getPulsarResources()); doReturn(authorizationService).when(brokerService).getAuthorizationService(); // lookup @@ -318,7 +318,7 @@ public void testVerifyOriginalPrincipalWithoutAuthDataForwardedFromProxy() throw @Test public void testVerifyAuthRoleAndAuthDataFromDirectConnectionBroker() throws Exception { - ServerCnx serverCnx = testPulsarServiceFactory.createServerCnxSpy(); + ServerCnx serverCnx = pulsarTestContext.createServerCnxSpy(); ChannelHandlerContext channelHandlerContext = mock(ChannelHandlerContext.class); Channel channel = mock(Channel.class); @@ -350,7 +350,7 @@ public void testVerifyAuthRoleAndAuthDataFromDirectConnectionBroker() throws Exc AuthorizationService authorizationService = spyWithClassAndConstructorArgsRecordingInvocations(AuthorizationService.class, svcConfig, - testPulsarServiceFactory.getPulsarResources()); + pulsarTestContext.getPulsarResources()); doReturn(authorizationService).when(brokerService).getAuthorizationService(); // lookup diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java index c4088d2e12489..b990ced12b141 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java @@ -71,7 +71,7 @@ import org.apache.bookkeeper.mledger.impl.PositionImpl; import org.apache.pulsar.broker.PulsarService; import org.apache.pulsar.broker.ServiceConfiguration; -import org.apache.pulsar.broker.TestPulsarService; +import org.apache.pulsar.broker.testcontext.PulsarTestContext; import org.apache.pulsar.broker.TransactionMetadataStoreService; import org.apache.pulsar.broker.auth.MockAuthenticationProvider; import org.apache.pulsar.broker.auth.MockMultiStageAuthenticationProvider; @@ -143,7 +143,7 @@ public class ServerCnxTest { private ServiceConfiguration svcConfig; private ServerCnx serverCnx; - protected TestPulsarService.Factory testPulsarServiceFactory; + protected PulsarTestContext pulsarTestContext; protected PulsarService pulsar; protected BrokerService brokerService; @@ -181,13 +181,13 @@ public void setup() throws Exception { svcConfig.setKeepAliveIntervalSeconds(inSec(1, TimeUnit.SECONDS)); svcConfig.setBacklogQuotaCheckEnabled(false); svcConfig.setClusterName("use"); - testPulsarServiceFactory = TestPulsarService.Factory.builder() + pulsarTestContext = PulsarTestContext.builder() .config(svcConfig) - .useSpies(true) + .spyByDefault() .build(); - pulsar = testPulsarServiceFactory.getPulsarService(); + pulsar = pulsarTestContext.getPulsarService(); - brokerService = testPulsarServiceFactory.getBrokerService(); + brokerService = pulsarTestContext.getBrokerService(); namespaceService = pulsar.getNamespaceService(); doReturn(CompletableFuture.completedFuture(null)).when(namespaceService).getBundleAsync(any()); @@ -217,9 +217,9 @@ public void teardown() throws Exception { if (channel != null) { channel.close(); } - if (testPulsarServiceFactory != null) { - testPulsarServiceFactory.close(); - testPulsarServiceFactory = null; + if (pulsarTestContext != null) { + pulsarTestContext.close(); + pulsarTestContext = null; } } @@ -895,7 +895,7 @@ public void testCreateProducerTimeout() throws Exception { () -> ((OpenLedgerCallback) invocationOnMock.getArguments()[2]).openLedgerComplete(ledgerMock, null)); return null; - }).when(testPulsarServiceFactory.getManagedLedgerFactory()) + }).when(pulsarTestContext.getManagedLedgerFactory()) .asyncOpen(matches(".*success.*"), any(ManagedLedgerConfig.class), any(OpenLedgerCallback.class), any(Supplier.class), any()); @@ -950,7 +950,7 @@ public void testCreateProducerTimeoutThenCreateSameNamedProducerShouldFail() thr () -> ((OpenLedgerCallback) invocationOnMock.getArguments()[2]).openLedgerComplete(ledgerMock, null)); return null; - }).when(testPulsarServiceFactory.getManagedLedgerFactory()) + }).when(pulsarTestContext.getManagedLedgerFactory()) .asyncOpen(matches(".*success.*"), any(ManagedLedgerConfig.class), any(OpenLedgerCallback.class), any(Supplier.class), any()); @@ -1017,7 +1017,7 @@ public void testCreateProducerMultipleTimeouts() throws Exception { ((OpenLedgerCallback) invocationOnMock.getArguments()[2]).openLedgerComplete(ledgerMock, null); return null; - }).when(testPulsarServiceFactory.getManagedLedgerFactory()) + }).when(pulsarTestContext.getManagedLedgerFactory()) .asyncOpen(matches(".*success.*"), any(ManagedLedgerConfig.class), any(OpenLedgerCallback.class), any(Supplier.class), any()); @@ -1096,7 +1096,7 @@ public void testCreateProducerBookieTimeout() throws Exception { () -> ((OpenLedgerCallback) invocationOnMock.getArguments()[2]).openLedgerComplete(ledgerMock, null)); return null; - }).when(testPulsarServiceFactory.getManagedLedgerFactory()) + }).when(pulsarTestContext.getManagedLedgerFactory()) .asyncOpen(matches(".*fail.*"), any(ManagedLedgerConfig.class), any(OpenLedgerCallback.class), any(Supplier.class), any()); @@ -1168,7 +1168,7 @@ public void testSubscribeTimeout() throws Exception { null)); return null; - }).when(testPulsarServiceFactory.getManagedLedgerFactory()) + }).when(pulsarTestContext.getManagedLedgerFactory()) .asyncOpen(matches(".*success.*"), any(ManagedLedgerConfig.class), any(OpenLedgerCallback.class), any(Supplier.class), any()); @@ -1243,7 +1243,7 @@ public void testSubscribeBookieTimeout() throws Exception { () -> ((OpenLedgerCallback) invocationOnMock.getArguments()[2]).openLedgerComplete(ledgerMock, null)); return null; - }).when(testPulsarServiceFactory.getManagedLedgerFactory()) + }).when(pulsarTestContext.getManagedLedgerFactory()) .asyncOpen(matches(".*success.*"), any(ManagedLedgerConfig.class), any(OpenLedgerCallback.class), any(Supplier.class), any()); @@ -1252,7 +1252,7 @@ public void testSubscribeBookieTimeout() throws Exception { openTopicFail.complete(() -> ((OpenLedgerCallback) invocationOnMock.getArguments()[2]) .openLedgerFailed(new ManagedLedgerException("Managed ledger failure"), null)); return null; - }).when(testPulsarServiceFactory.getManagedLedgerFactory()) + }).when(pulsarTestContext.getManagedLedgerFactory()) .asyncOpen(matches(".*fail.*"), any(ManagedLedgerConfig.class), any(OpenLedgerCallback.class), any(Supplier.class), any()); @@ -1494,7 +1494,7 @@ public void testProducerSuccessOnEncryptionRequiredTopic() throws Exception { // add `clusterDispatchRate` otherwise there will be a NPE // `org.apache.pulsar.broker.service.AbstractTopic.updateNamespaceReplicatorDispatchRate` policies.replicatorDispatchRate = new HashMap<>(); - testPulsarServiceFactory.getPulsarResources().getNamespaceResources() + pulsarTestContext.getPulsarResources().getNamespaceResources() .createPolicies(TopicName.get(encryptionRequiredTopicName).getNamespaceObject(), policies); @@ -1532,7 +1532,7 @@ public void testProducerFailureOnEncryptionRequiredTopic() throws Exception { // add `clusterDispatchRate` otherwise there will be a NPE // `org.apache.pulsar.broker.service.AbstractTopic.updateNamespaceReplicatorDispatchRate` policies.replicatorDispatchRate = new HashMap<>(); - testPulsarServiceFactory.getPulsarResources().getNamespaceResources() + pulsarTestContext.getPulsarResources().getNamespaceResources() .createPolicies(TopicName.get(encryptionRequiredTopicName).getNamespaceObject(), policies); @@ -1574,7 +1574,7 @@ public void testProducerFailureOnEncryptionRequiredOnBroker() throws Exception { // add `clusterDispatchRate` otherwise there will be a NPE // `org.apache.pulsar.broker.service.AbstractTopic.updateNamespaceReplicatorDispatchRate` policies.replicatorDispatchRate = new HashMap<>(); - testPulsarServiceFactory.getPulsarResources().getNamespaceResources() + pulsarTestContext.getPulsarResources().getNamespaceResources() .createPolicies(TopicName.get(encryptionRequiredTopicName).getNamespaceObject(), policies); @@ -1614,7 +1614,7 @@ public void testSendSuccessOnEncryptionRequiredTopic() throws Exception { // add `clusterDispatchRate` otherwise there will be a NPE // `org.apache.pulsar.broker.service.AbstractTopic.updateNamespaceReplicatorDispatchRate` policies.replicatorDispatchRate = new HashMap<>(); - testPulsarServiceFactory.getPulsarResources().getNamespaceResources() + pulsarTestContext.getPulsarResources().getNamespaceResources() .createPolicies(TopicName.get(encryptionRequiredTopicName).getNamespaceObject(), policies); @@ -1660,7 +1660,7 @@ public void testSendFailureOnEncryptionRequiredTopic() throws Exception { // add `clusterDispatchRate` otherwise there will be a NPE // `org.apache.pulsar.broker.service.AbstractTopic.updateNamespaceReplicatorDispatchRate` policies.replicatorDispatchRate = new HashMap<>(); - testPulsarServiceFactory.getPulsarResources().getNamespaceResources() + pulsarTestContext.getPulsarResources().getNamespaceResources() .createPolicies(TopicName.get(encryptionRequiredTopicName).getNamespaceObject(), policies); @@ -1740,7 +1740,7 @@ private void setupMLAsyncCallbackMocks() { Thread.sleep(300); ((OpenLedgerCallback) invocationOnMock.getArguments()[2]).openLedgerComplete(ledgerMock, null); return null; - }).when(testPulsarServiceFactory.getManagedLedgerFactory()) + }).when(pulsarTestContext.getManagedLedgerFactory()) .asyncOpen(matches(".*success.*"), any(ManagedLedgerConfig.class), any(OpenLedgerCallback.class), any(Supplier.class), any()); @@ -1751,7 +1751,7 @@ private void setupMLAsyncCallbackMocks() { .openLedgerFailed(new ManagedLedgerException("Managed ledger failure"), null)).start(); return null; - }).when(testPulsarServiceFactory.getManagedLedgerFactory()) + }).when(pulsarTestContext.getManagedLedgerFactory()) .asyncOpen(matches(".*fail.*"), any(ManagedLedgerConfig.class), any(OpenLedgerCallback.class), any(Supplier.class), any()); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherFailoverConsumerStreamingDispatcherTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherFailoverConsumerStreamingDispatcherTest.java index 00e31c51c9ca5..f86fe0701dc70 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherFailoverConsumerStreamingDispatcherTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherFailoverConsumerStreamingDispatcherTest.java @@ -32,6 +32,6 @@ public class PersistentDispatcherFailoverConsumerStreamingDispatcherTest extends @BeforeMethod(alwaysRun = true) public void setup() throws Exception { super.setup(); - testPulsarServiceFactory.getConfig().setStreamingDispatch(true); + pulsarTestContext.getConfig().setStreamingDispatch(true); } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicStreamingDispatcherTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicStreamingDispatcherTest.java index d7699e0b8175a..440cbbe290c4d 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicStreamingDispatcherTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicStreamingDispatcherTest.java @@ -33,7 +33,7 @@ public class PersistentTopicStreamingDispatcherTest extends PersistentTopicTest @BeforeMethod(alwaysRun = true) public void setup() throws Exception { super.setup(); - ServiceConfiguration config = testPulsarServiceFactory.getConfig(); + ServiceConfiguration config = pulsarTestContext.getConfig(); config.setTopicLevelPoliciesEnabled(false); config.setSystemTopicEnabled(false); config.setStreamingDispatch(true); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/AbstractTestPulsarService.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/AbstractTestPulsarService.java new file mode 100644 index 0000000000000..e8a9063a1ea68 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/AbstractTestPulsarService.java @@ -0,0 +1,88 @@ +/* + * 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.broker.testcontext; + +import org.apache.pulsar.broker.BookKeeperClientFactory; +import org.apache.pulsar.broker.PulsarServerException; +import org.apache.pulsar.broker.PulsarService; +import org.apache.pulsar.broker.ServiceConfiguration; +import org.apache.pulsar.broker.intercept.BrokerInterceptor; +import org.apache.pulsar.broker.service.PulsarMetadataEventSynchronizer; +import org.apache.pulsar.compaction.Compactor; +import org.apache.pulsar.metadata.api.MetadataStore; +import org.apache.pulsar.metadata.api.MetadataStoreException; +import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended; + +abstract class AbstractTestPulsarService extends PulsarService { + protected final MetadataStoreExtended localMetadataStore; + protected final MetadataStoreExtended configurationMetadataStore; + protected final Compactor compactor; + protected final BrokerInterceptor brokerInterceptor; + protected final BookKeeperClientFactory bookKeeperClientFactory; + + public AbstractTestPulsarService(ServiceConfiguration config, MetadataStoreExtended localMetadataStore, + MetadataStoreExtended configurationMetadataStore, Compactor compactor, + BrokerInterceptor brokerInterceptor, + BookKeeperClientFactory bookKeeperClientFactory) { + super(config); + this.localMetadataStore = + NonClosingProxyHandler.createNonClosingProxy(localMetadataStore, MetadataStoreExtended.class); + this.configurationMetadataStore = + NonClosingProxyHandler.createNonClosingProxy(configurationMetadataStore, MetadataStoreExtended.class); + this.compactor = compactor; + this.brokerInterceptor = brokerInterceptor; + this.bookKeeperClientFactory = bookKeeperClientFactory; + } + + @Override + public MetadataStore createConfigurationMetadataStore(PulsarMetadataEventSynchronizer synchronizer) + throws MetadataStoreException { + + return configurationMetadataStore; + } + + @Override + public MetadataStoreExtended createLocalMetadataStore(PulsarMetadataEventSynchronizer synchronizer) + throws MetadataStoreException, PulsarServerException { + return localMetadataStore; + } + + @Override + public Compactor getCompactor() throws PulsarServerException { + if (compactor != null) { + return compactor; + } else { + return super.getCompactor(); + } + } + + @Override + public BrokerInterceptor getBrokerInterceptor() { + if (brokerInterceptor != null) { + return brokerInterceptor; + } else { + return super.getBrokerInterceptor(); + } + } + + @Override + public BookKeeperClientFactory newBookKeeperClientFactory() { + return bookKeeperClientFactory; + } +} diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/MockBookKeeperClientFactory.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/MockBookKeeperClientFactory.java new file mode 100644 index 0000000000000..c173be9046b3f --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/MockBookKeeperClientFactory.java @@ -0,0 +1,60 @@ +/* + * 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.broker.testcontext; + +import io.netty.channel.EventLoopGroup; +import java.util.Map; +import java.util.Optional; +import org.apache.bookkeeper.client.BookKeeper; +import org.apache.bookkeeper.client.EnsemblePlacementPolicy; +import org.apache.bookkeeper.stats.StatsLogger; +import org.apache.pulsar.broker.BookKeeperClientFactory; +import org.apache.pulsar.broker.ServiceConfiguration; +import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended; + +class MockBookKeeperClientFactory implements BookKeeperClientFactory { + private final BookKeeper mockBookKeeper; + + MockBookKeeperClientFactory(BookKeeper mockBookKeeper) { + this.mockBookKeeper = mockBookKeeper; + } + + @Override + public BookKeeper create(ServiceConfiguration conf, MetadataStoreExtended store, + EventLoopGroup eventLoopGroup, + Optional> ensemblePlacementPolicyClass, + Map properties) { + // Always return the same instance (so that we don't loose the mock BK content on broker restart + return mockBookKeeper; + } + + @Override + public BookKeeper create(ServiceConfiguration conf, MetadataStoreExtended store, + EventLoopGroup eventLoopGroup, + Optional> ensemblePlacementPolicyClass, + Map properties, StatsLogger statsLogger) { + // Always return the same instance (so that we don't loose the mock BK content on broker restart + return mockBookKeeper; + } + + @Override + public void close() { + // no-op + } +} diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/NonClosableMockBookKeeper.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/NonClosableMockBookKeeper.java new file mode 100644 index 0000000000000..23cc07d3bdfd7 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/NonClosableMockBookKeeper.java @@ -0,0 +1,44 @@ +/* + * 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.broker.testcontext; + +import org.apache.bookkeeper.client.PulsarMockBookKeeper; +import org.apache.bookkeeper.common.util.OrderedExecutor; + +// Prevent the MockBookKeeper instance from being closed when the broker is restarted within a test +class NonClosableMockBookKeeper extends PulsarMockBookKeeper { + + public NonClosableMockBookKeeper(OrderedExecutor executor) throws Exception { + super(executor); + } + + @Override + public void close() { + // no-op + } + + @Override + public void shutdown() { + // no-op + } + + public void reallyShutdown() { + super.shutdown(); + } +} diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/NonClosingProxyHandler.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/NonClosingProxyHandler.java new file mode 100644 index 0000000000000..3156d25a74e9d --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/NonClosingProxyHandler.java @@ -0,0 +1,73 @@ +/* + * 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.broker.testcontext; + +import java.lang.reflect.InvocationHandler; +import java.lang.reflect.Method; +import java.lang.reflect.Proxy; + +public class NonClosingProxyHandler implements InvocationHandler { + private final AutoCloseable delegate; + + NonClosingProxyHandler(AutoCloseable delegate) { + this.delegate = delegate; + } + + public static T createNonClosingProxy(T delegate, Class interfaceClass) { + if (isNonClosingProxy(delegate)) { + return delegate; + } + return interfaceClass.cast(Proxy.newProxyInstance(delegate.getClass().getClassLoader(), + new Class[] {interfaceClass}, new NonClosingProxyHandler(delegate))); + } + + public static boolean isNonClosingProxy(Object instance) { + return Proxy.isProxyClass(instance.getClass()) + && Proxy.getInvocationHandler(instance) instanceof NonClosingProxyHandler; + } + + public static I getDelegate(T instance) { + if (isNonClosingProxy(instance)) { + return (T) ((NonClosingProxyHandler) Proxy.getInvocationHandler(instance)).getDelegate(); + } else { + throw new IllegalArgumentException("not a proxy instance with NonClosingProxyHandler"); + } + } + + public static void reallyClose(T instance) throws Exception { + if (isNonClosingProxy(instance)) { + getDelegate(instance).close(); + } else { + instance.close(); + } + } + + public AutoCloseable getDelegate() { + return delegate; + } + + @Override + public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { + if (method.getName().equals("close")) { + return null; + } else { + return method.invoke(delegate, args); + } + } +} diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/NonStartableTestPulsarService.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/NonStartableTestPulsarService.java new file mode 100644 index 0000000000000..c65798ed7f6ed --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/NonStartableTestPulsarService.java @@ -0,0 +1,196 @@ +/* + * 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.broker.testcontext; + +import static org.apache.pulsar.broker.BrokerTestUtil.spyWithClassAndConstructorArgs; +import static org.mockito.Mockito.mock; +import io.netty.channel.EventLoopGroup; +import java.util.Collections; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.function.Supplier; +import org.apache.pulsar.broker.BookKeeperClientFactory; +import org.apache.pulsar.broker.PulsarServerException; +import org.apache.pulsar.broker.PulsarService; +import org.apache.pulsar.broker.ServiceConfiguration; +import org.apache.pulsar.broker.intercept.BrokerInterceptor; +import org.apache.pulsar.broker.namespace.NamespaceService; +import org.apache.pulsar.broker.resources.NamespaceResources; +import org.apache.pulsar.broker.resources.PulsarResources; +import org.apache.pulsar.broker.resources.TopicResources; +import org.apache.pulsar.broker.service.BrokerService; +import org.apache.pulsar.broker.service.schema.DefaultSchemaRegistryService; +import org.apache.pulsar.broker.service.schema.SchemaRegistryService; +import org.apache.pulsar.broker.storage.ManagedLedgerStorage; +import org.apache.pulsar.client.api.PulsarClient; +import org.apache.pulsar.client.api.PulsarClientException; +import org.apache.pulsar.client.impl.PulsarClientImpl; +import org.apache.pulsar.client.impl.conf.ClientConfigurationData; +import org.apache.pulsar.common.naming.TopicName; +import org.apache.pulsar.compaction.Compactor; +import org.apache.pulsar.metadata.api.MetadataStore; +import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended; + +/** + * Subclass of PulsarService that is used for some tests. + * This was written as a replacement for the previous Mockito Spy over PulsarService solution which caused + * a flaky test issue https://github.com/apache/pulsar/issues/13620. + */ + +class NonStartableTestPulsarService extends AbstractTestPulsarService { + private final PulsarResources pulsarResources; + private final ManagedLedgerStorage managedLedgerClientFactory; + private final BrokerService brokerService; + + private final SchemaRegistryService schemaRegistryService; + + private final PulsarClientImpl pulsarClient; + + private final NamespaceService namespaceService; + + public NonStartableTestPulsarService(SpyConfig spyConfig, ServiceConfiguration config, + MetadataStoreExtended localMetadataStore, + MetadataStoreExtended configurationMetadataStore, + Compactor compactor, BrokerInterceptor brokerInterceptor, + BookKeeperClientFactory bookKeeperClientFactory, + PulsarResources pulsarResources, + ManagedLedgerStorage managedLedgerClientFactory) { + super(config, localMetadataStore, configurationMetadataStore, compactor, brokerInterceptor, + bookKeeperClientFactory); + this.pulsarResources = pulsarResources; + this.managedLedgerClientFactory = managedLedgerClientFactory; + try { + this.brokerService = spyConfig.getBrokerService().spy(TestBrokerService.class, this, getIoEventLoopGroup()); + } catch (Exception e) { + throw new RuntimeException(e); + } + this.schemaRegistryService = spyWithClassAndConstructorArgs(DefaultSchemaRegistryService.class); + this.pulsarClient = mock(PulsarClientImpl.class); + this.namespaceService = mock(NamespaceService.class); + try { + startNamespaceService(); + } catch (PulsarServerException e) { + throw new RuntimeException(e); + } + } + + @Override + public void start() throws PulsarServerException { + throw new UnsupportedOperationException("Cannot start a non-startable TestPulsarService"); + } + + @Override + public Supplier getNamespaceServiceProvider() throws PulsarServerException { + return () -> namespaceService; + } + + @Override + public synchronized PulsarClient getClient() throws PulsarServerException { + return pulsarClient; + } + + @Override + public PulsarClientImpl createClientImpl(ClientConfigurationData clientConf) throws PulsarClientException { + return pulsarClient; + } + + @Override + public SchemaRegistryService getSchemaRegistryService() { + return schemaRegistryService; + } + + @Override + public PulsarResources getPulsarResources() { + return pulsarResources; + } + + public BrokerService getBrokerService() { + return brokerService; + } + + @Override + public MetadataStore getConfigurationMetadataStore() { + return configurationMetadataStore; + } + + @Override + public MetadataStoreExtended getLocalMetadataStore() { + return localMetadataStore; + } + + @Override + public ManagedLedgerStorage getManagedLedgerClientFactory() { + return managedLedgerClientFactory; + } + + @Override + protected PulsarResources newPulsarResources() { + return pulsarResources; + } + + @Override + protected ManagedLedgerStorage newManagedLedgerClientFactory() throws Exception { + return managedLedgerClientFactory; + } + + @Override + protected BrokerService newBrokerService(PulsarService pulsar) throws Exception { + return brokerService; + } + + @Override + public BookKeeperClientFactory getBookKeeperClientFactory() { + return bookKeeperClientFactory; + } + + static class TestBrokerService extends BrokerService { + + TestBrokerService(PulsarService pulsar, EventLoopGroup eventLoopGroup) throws Exception { + super(pulsar, eventLoopGroup); + } + + @Override + protected CompletableFuture> fetchTopicPropertiesAsync(TopicName topicName) { + return CompletableFuture.completedFuture(Collections.emptyMap()); + } + } + + static class TestPulsarResources extends PulsarResources { + + private final TopicResources topicResources; + private final NamespaceResources namespaceResources; + + public TestPulsarResources(MetadataStore localMetadataStore, MetadataStore configurationMetadataStore, + TopicResources topicResources, NamespaceResources namespaceResources) { + super(localMetadataStore, configurationMetadataStore); + this.topicResources = topicResources; + this.namespaceResources = namespaceResources; + } + + @Override + public TopicResources getTopicResources() { + return topicResources; + } + + @Override + public NamespaceResources getNamespaceResources() { + return namespaceResources; + } + } +} diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/PulsarTestContext.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/PulsarTestContext.java new file mode 100644 index 0000000000000..f42c3cb146edb --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/PulsarTestContext.java @@ -0,0 +1,411 @@ +/* + * 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.broker.testcontext; + +import static org.apache.pulsar.broker.BrokerTestUtil.spyWithClassAndConstructorArgsRecordingInvocations; +import static org.mockito.Mockito.mock; +import com.google.common.util.concurrent.MoreExecutors; +import io.netty.channel.EventLoopGroup; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; +import java.util.function.Consumer; +import java.util.function.Function; +import lombok.AccessLevel; +import lombok.Builder; +import lombok.Getter; +import lombok.Singular; +import lombok.ToString; +import lombok.extern.slf4j.Slf4j; +import org.apache.bookkeeper.client.BookKeeper; +import org.apache.bookkeeper.common.util.OrderedExecutor; +import org.apache.bookkeeper.mledger.ManagedLedgerFactory; +import org.apache.bookkeeper.stats.NullStatsProvider; +import org.apache.bookkeeper.stats.StatsProvider; +import org.apache.bookkeeper.util.ZkUtils; +import org.apache.pulsar.broker.BookKeeperClientFactory; +import org.apache.pulsar.broker.PulsarService; +import org.apache.pulsar.broker.ServiceConfiguration; +import org.apache.pulsar.broker.intercept.BrokerInterceptor; +import org.apache.pulsar.broker.resources.NamespaceResources; +import org.apache.pulsar.broker.resources.PulsarResources; +import org.apache.pulsar.broker.resources.TopicResources; +import org.apache.pulsar.broker.service.BrokerService; +import org.apache.pulsar.broker.service.ServerCnx; +import org.apache.pulsar.broker.storage.ManagedLedgerStorage; +import org.apache.pulsar.common.util.GracefulExecutorServicesShutdown; +import org.apache.pulsar.compaction.Compactor; +import org.apache.pulsar.metadata.api.MetadataStore; +import org.apache.pulsar.metadata.api.MetadataStoreConfig; +import org.apache.pulsar.metadata.api.MetadataStoreException; +import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended; +import org.apache.pulsar.metadata.impl.MetadataStoreFactoryImpl; +import org.apache.pulsar.metadata.impl.ZKMetadataStore; +import org.apache.zookeeper.CreateMode; +import org.apache.zookeeper.MockZooKeeper; +import org.apache.zookeeper.MockZooKeeperSession; +import org.apache.zookeeper.data.ACL; +import org.jetbrains.annotations.NotNull; + +@Slf4j +@ToString +@Getter +@Builder(builderClassName = "Builder") +public class PulsarTestContext implements AutoCloseable { + private final ServiceConfiguration config; + private final MetadataStoreExtended localMetadataStore; + private final MetadataStoreExtended configurationMetadataStore; + private final PulsarResources pulsarResources; + + private final OrderedExecutor executor; + + private final ManagedLedgerStorage managedLedgerClientFactory; + + private final PulsarService pulsarService; + + private final Compactor compactor; + + private final BrokerService brokerService; + + @Getter(AccessLevel.NONE) + @Singular("registerCloseable") + private final List closeables; + + private final BrokerInterceptor brokerInterceptor; + + private final BookKeeper bookKeeperClient; + + private final boolean startable; + + + public ManagedLedgerFactory getManagedLedgerFactory() { + return managedLedgerClientFactory.getManagedLedgerFactory(); + } + + public static Builder startableBuilder() { + return new StartableCustomBuilder(); + } + + public static Builder builder() { + return new NonStartableCustomBuilder(); + } + + public void close() throws Exception { + for (int i = closeables.size() - 1; i >= 0; i--) { + try { + closeables.get(i).close(); + } catch (Exception e) { + log.error("Failure in calling cleanup function", e); + } + } + } + + public ServerCnx createServerCnxSpy() { + return spyWithClassAndConstructorArgsRecordingInvocations(ServerCnx.class, + getPulsarService()); + } + + public static class Builder { + protected boolean useTestPulsarResources = false; + protected MetadataStore pulsarResourcesMetadataStore; + protected Function brokerServiceFunction; + protected SpyConfig.Builder spyConfigBuilder = SpyConfig.builder(SpyConfig.SpyType.NONE); + + public Builder spyByDefault() { + spyConfigBuilder = SpyConfig.builder(SpyConfig.SpyType.SPY); + return this; + } + + public Builder spyConfig(Consumer spyConfigCustomizer) { + spyConfigCustomizer.accept(spyConfigBuilder); + return this; + } + + + public Builder reuseMockBookkeeperAndMetadataStores(PulsarTestContext otherContext) { + bookKeeperClient(otherContext.getBookKeeperClient()); + localMetadataStore(NonClosingProxyHandler.createNonClosingProxy(otherContext.getLocalMetadataStore(), + MetadataStoreExtended.class + )); + configurationMetadataStore(NonClosingProxyHandler.createNonClosingProxy( + otherContext.getConfigurationMetadataStore(), MetadataStoreExtended.class + )); + return this; + } + + public Builder withMockZookeeper() { + try { + MockZooKeeper mockZooKeeper = createMockZooKeeper(); + registerCloseable(mockZooKeeper::shutdown); + MockZooKeeperSession mockZooKeeperSession = MockZooKeeperSession.newInstance(mockZooKeeper); + ZKMetadataStore zkMetadataStore = new ZKMetadataStore(mockZooKeeperSession); + registerCloseable(zkMetadataStore::close); + localMetadataStore(zkMetadataStore); + configurationMetadataStore(zkMetadataStore); + } catch (Exception e) { + throw new RuntimeException(e); + } + return this; + } + + private static MockZooKeeper createMockZooKeeper() throws Exception { + MockZooKeeper zk = MockZooKeeper.newInstance(MoreExecutors.newDirectExecutorService()); + List dummyAclList = new ArrayList<>(0); + + ZkUtils.createFullPathOptimistic(zk, "/ledgers/available/192.168.1.1:" + 5000, + "".getBytes(StandardCharsets.UTF_8), dummyAclList, CreateMode.PERSISTENT); + + zk.create("/ledgers/LAYOUT", "1\nflat:1".getBytes(StandardCharsets.UTF_8), dummyAclList, + CreateMode.PERSISTENT); + return zk; + } + + public Builder useTestPulsarResources() { + if (startable) { + throw new IllegalStateException("Cannot useTestPulsarResources when startable."); + } + useTestPulsarResources = true; + return this; + } + + public Builder useTestPulsarResources(MetadataStore metadataStore) { + if (startable) { + throw new IllegalStateException("Cannot useTestPulsarResources when startable."); + } + useTestPulsarResources = true; + pulsarResourcesMetadataStore = metadataStore; + return this; + } + + public Builder managedLedgerClients(BookKeeper bookKeeperClient, + ManagedLedgerFactory managedLedgerFactory) { + return managedLedgerClientFactory( + PulsarTestContext.createManagedLedgerClientFactory(bookKeeperClient, managedLedgerFactory)); + } + + public Builder brokerServiceFunction( + Function brokerServiceFunction) { + this.brokerServiceFunction = brokerServiceFunction; + return this; + } + } + + static abstract class AbstractCustomBuilder extends Builder { + AbstractCustomBuilder(boolean startable) { + super.startable = startable; + } + + public Builder startable(boolean startable) { + throw new UnsupportedOperationException("Cannot change startability after builder creation."); + } + + @Override + public final PulsarTestContext build() { + SpyConfig spyConfig = spyConfigBuilder.build(); + if (super.config == null) { + ServiceConfiguration svcConfig = new ServiceConfiguration(); + initializeConfig(svcConfig); + config(svcConfig); + } + initializeCommonPulsarServices(spyConfig); + initializePulsarServices(spyConfig, this); + if (super.startable) { + try { + super.pulsarService.start(); + } catch (Exception e) { + throw new RuntimeException(e); + } + } + brokerService(super.pulsarService.getBrokerService()); + return super.build(); + } + + protected void initializeConfig(ServiceConfiguration svcConfig) { + svcConfig.setBrokerShutdownTimeoutMs(0L); + svcConfig.setLoadBalancerOverrideBrokerNicSpeedGbps(Optional.of(1.0d)); + svcConfig.setClusterName("pulsar-cluster"); + svcConfig.setNumIOThreads(1); + svcConfig.setNumOrderedExecutorThreads(1); + svcConfig.setNumExecutorThreadPoolSize(2); + svcConfig.setNumCacheExecutorThreadPoolSize(2); + svcConfig.setNumHttpServerThreads(2); + } + + private void initializeCommonPulsarServices(SpyConfig spyConfig) { + if (super.bookKeeperClient == null && super.managedLedgerClientFactory == null) { + if (super.executor == null) { + OrderedExecutor createdExecutor = OrderedExecutor.newBuilder().numThreads(1) + .name(NonStartableTestPulsarService.class.getSimpleName() + "-executor").build(); + registerCloseable(() -> GracefulExecutorServicesShutdown.initiate() + .timeout(Duration.ZERO) + .shutdown(createdExecutor) + .handle().get()); + super.executor = createdExecutor; + } + NonClosableMockBookKeeper mockBookKeeper; + try { + mockBookKeeper = + spyConfig.getBookKeeperClient().spy(NonClosableMockBookKeeper.class, super.executor); + } catch (Exception e) { + throw new RuntimeException(e); + } + registerCloseable(mockBookKeeper::reallyShutdown); + bookKeeperClient(mockBookKeeper); + } + if (super.bookKeeperClient == null && super.managedLedgerClientFactory != null) { + bookKeeperClient(super.managedLedgerClientFactory.getBookKeeperClient()); + } + if (super.localMetadataStore == null || super.configurationMetadataStore == null) { + try { + MetadataStoreExtended store = MetadataStoreFactoryImpl.createExtended("memory:local", + MetadataStoreConfig.builder().build()); + registerCloseable(store::close); + if (super.localMetadataStore == null) { + localMetadataStore(store); + } + if (super.configurationMetadataStore == null) { + configurationMetadataStore(store); + } + } catch (MetadataStoreException e) { + throw new RuntimeException(e); + } + } + } + + protected abstract void initializePulsarServices(SpyConfig spyConfig, Builder builder); + } + + + static class StartableCustomBuilder extends AbstractCustomBuilder { + StartableCustomBuilder() { + super(true); + } + + @Override + public Builder managedLedgerClientFactory(ManagedLedgerStorage managedLedgerClientFactory) { + throw new IllegalStateException("Cannot set managedLedgerClientFactory when startable."); + } + + @Override + public Builder pulsarResources(PulsarResources pulsarResources) { + throw new IllegalStateException("Cannot set pulsarResources when startable."); + } + + @Override + protected void initializePulsarServices(SpyConfig spyConfig, Builder builder) { + BookKeeperClientFactory bookKeeperClientFactory = + new MockBookKeeperClientFactory(builder.bookKeeperClient); + PulsarService pulsarService = spyConfig.getPulsarBroker() + .spy(StartableTestPulsarService.class, builder.config, builder.localMetadataStore, + builder.configurationMetadataStore, builder.compactor, builder.brokerInterceptor, + bookKeeperClientFactory); + registerCloseable(pulsarService::close); + pulsarService(pulsarService); + } + + @Override + protected void initializeConfig(ServiceConfiguration svcConfig) { + super.initializeConfig(svcConfig); + svcConfig.setBrokerShutdownTimeoutMs(5000L); + } + } + + static class NonStartableCustomBuilder extends AbstractCustomBuilder { + + NonStartableCustomBuilder() { + super(false); + } + + @Override + protected void initializePulsarServices(SpyConfig spyConfig, Builder builder) { + if (builder.managedLedgerClientFactory == null) { + ManagedLedgerFactory mlFactoryMock = mock(ManagedLedgerFactory.class); + managedLedgerClientFactory( + PulsarTestContext.createManagedLedgerClientFactory(builder.bookKeeperClient, mlFactoryMock)); + } + if (builder.pulsarResources == null) { + SpyConfig.SpyType spyConfigPulsarResources = spyConfig.getPulsarResources(); + if (useTestPulsarResources) { + MetadataStore metadataStore = pulsarResourcesMetadataStore; + if (metadataStore == null) { + metadataStore = builder.configurationMetadataStore; + } + NamespaceResources nsr = spyConfigPulsarResources.spy(NamespaceResources.class, metadataStore, 30); + TopicResources tsr = spyConfigPulsarResources.spy(TopicResources.class, metadataStore); + pulsarResources( + spyConfigPulsarResources.spy( + NonStartableTestPulsarService.TestPulsarResources.class, builder.localMetadataStore, + builder.configurationMetadataStore, + tsr, nsr)); + } else { + pulsarResources( + spyConfigPulsarResources.spy(PulsarResources.class, builder.localMetadataStore, + builder.configurationMetadataStore)); + } + } + BookKeeperClientFactory bookKeeperClientFactory = + new MockBookKeeperClientFactory(builder.bookKeeperClient); + PulsarService pulsarService = spyConfig.getPulsarBroker() + .spy(NonStartableTestPulsarService.class, spyConfig, builder.config, builder.localMetadataStore, + builder.configurationMetadataStore, builder.compactor, builder.brokerInterceptor, + bookKeeperClientFactory, builder.pulsarResources, + builder.managedLedgerClientFactory); + registerCloseable(pulsarService::close); + pulsarService(pulsarService); + } + } + + @NotNull + private static ManagedLedgerStorage createManagedLedgerClientFactory(BookKeeper bookKeeperClient, + ManagedLedgerFactory managedLedgerFactory) { + return new ManagedLedgerStorage() { + + @Override + public void initialize(ServiceConfiguration conf, MetadataStoreExtended metadataStore, + BookKeeperClientFactory bookkeeperProvider, EventLoopGroup eventLoopGroup) + throws Exception { + + } + + @Override + public ManagedLedgerFactory getManagedLedgerFactory() { + return managedLedgerFactory; + } + + @Override + public StatsProvider getStatsProvider() { + return new NullStatsProvider(); + } + + @Override + public BookKeeper getBookKeeperClient() { + return bookKeeperClient; + } + + @Override + public void close() throws IOException { + + } + }; + } +} diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/SpyConfig.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/SpyConfig.java new file mode 100644 index 0000000000000..2e3b863cc8aaf --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/SpyConfig.java @@ -0,0 +1,83 @@ +/* + * 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.broker.testcontext; + +import lombok.Value; +import org.apache.pulsar.broker.BrokerTestUtil; +import org.mockito.Mockito; +import org.mockito.internal.creation.instance.ConstructorInstantiator; + +@lombok.Builder(builderClassName = "Builder", toBuilder = true) +@Value +public class SpyConfig { + public enum SpyType { + NONE, + SPY, + SPY_ALSO_INVOCATIONS; + + public T spy(T object) { + if (object == null) { + return null; + } + switch (this) { + case NONE: + return object; + case SPY: + return BrokerTestUtil.spyWithoutRecordingInvocations(object); + case SPY_ALSO_INVOCATIONS: + return Mockito.spy(object); + default: + throw new UnsupportedOperationException("Unknown spy type: " + this); + } + } + + public T spy(Class clazz, Object... args) { + switch (this) { + case NONE: + // Use Mockito's internal class to instantiate the object + return new ConstructorInstantiator(false, args).newInstance(clazz); + case SPY: + return BrokerTestUtil.spyWithClassAndConstructorArgs(clazz, args); + case SPY_ALSO_INVOCATIONS: + return BrokerTestUtil.spyWithClassAndConstructorArgsRecordingInvocations(clazz, args); + default: + throw new UnsupportedOperationException("Unknown spy type: " + this); + } + } + } + + private final SpyType pulsarBroker; + private final SpyType pulsarResources; + private final SpyType brokerService; + private final SpyType bookKeeperClient; + + public static Builder builder() { + return builder(SpyType.NONE); + } + + public static Builder builder(SpyType defaultSpyType) { + Builder spyConfigBuilder = new Builder(); + spyConfigBuilder.pulsarBroker(defaultSpyType); + spyConfigBuilder.pulsarResources(defaultSpyType); + spyConfigBuilder.brokerService(defaultSpyType); + spyConfigBuilder.bookKeeperClient(defaultSpyType); + return spyConfigBuilder; + } +} diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/StartableTestPulsarService.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/StartableTestPulsarService.java new file mode 100644 index 0000000000000..a171ce97c11e0 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/StartableTestPulsarService.java @@ -0,0 +1,37 @@ +/* + * 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.broker.testcontext; + +import org.apache.pulsar.broker.BookKeeperClientFactory; +import org.apache.pulsar.broker.ServiceConfiguration; +import org.apache.pulsar.broker.intercept.BrokerInterceptor; +import org.apache.pulsar.compaction.Compactor; +import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended; + +class StartableTestPulsarService extends AbstractTestPulsarService { + public StartableTestPulsarService(ServiceConfiguration config, + MetadataStoreExtended localMetadataStore, + MetadataStoreExtended configurationMetadataStore, + Compactor compactor, + BrokerInterceptor brokerInterceptor, + BookKeeperClientFactory bookKeeperClientFactory) { + super(config, localMetadataStore, configurationMetadataStore, compactor, brokerInterceptor, + bookKeeperClientFactory); + } +}