Skip to content

[improve][test] Refactor TestPulsarService to PulsarTestContext and add support for starting #19337

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
lhotari merged 1 commit into from
Jan 30, 2023
Merged

[improve][test] Refactor TestPulsarService to PulsarTestContext and add support for starting #19337

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
501 changes: 0 additions & 501 deletions pulsar-broker/src/test/java/org/apache/pulsar/broker/TestPulsarService.java
View file
Open in desktop

This file was deleted.

180 changes: 22 additions & 158 deletions ...r/src/test/java/org/apache/pulsar/broker/namespace/OwnerShipForCurrentServerTestBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<SameThreadOrderedSafeExecutor> orderedExecutorList = new ArrayList<>();
@Getter
private final List<ServiceConfiguration> serviceConfigurationList = new ArrayList<>();
@Getter
Expand All @@ -72,9 +49,7 @@ public class OwnerShipForCurrentServerTestBase {
protected PulsarAdmin admin;
protected PulsarClient pulsarClient;

private MockZooKeeper mockZooKeeper;
private OrderedExecutor bkExecutor;
private NonClosableMockBookKeeper mockBookKeeper;
protected List<PulsarTestContext> pulsarTestContexts = new ArrayList<>();

public void internalSetup() throws Exception {
init();
Expand All @@ -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();
}

Expand All @@ -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));

@michaeljmarshall michaeljmarshall Jan 30, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why are we getting the first one here?

@lhotari lhotari Jan 30, 2023

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

PulsarTestContext holds the mock bookkeeper and metadata store (mockzookeeper) references. The "reuseMockBookkeeperAndMetadataStores" can be used to share the mock bookkeeper and zookeeper from another context.

} 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<NamespaceService> 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<ACL> 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<Class<? extends EnsemblePlacementPolicy>> ensemblePlacementPolicyClass,
Map<String, Object> 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<Class<? extends EnsemblePlacementPolicy>> ensemblePlacementPolicyClass,
Map<String, Object> 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
Expand All @@ -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);
}
}

}
16 changes: 8 additions & 8 deletions pulsar-broker/src/test/java/org/apache/pulsar/broker/service/MessageCumulativeAckTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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();
Expand All @@ -74,17 +74,17 @@ 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());
}

@AfterMethod(alwaysRun = true)
public void shutdown() throws Exception {
if (testPulsarServiceFactory != null) {
testPulsarServiceFactory.close();
testPulsarServiceFactory = null;
if (pulsarTestContext != null) {
pulsarTestContext.close();
pulsarTestContext = null;
}
sub = null;
}
Expand Down
Loading