Skip to content

Commit

Permalink
fix tests
Browse files Browse the repository at this point in the history
  • Loading branch information
@poorbarcode
poorbarcode committed Apr 22, 2024
1 parent 0390ecf commit 5793ca1
Show file tree
Hide file tree
Showing 2 changed files with 92 additions and 23 deletions.
77 changes: 54 additions & 23 deletions pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,9 +21,9 @@
import static org.mockito.Mockito.any;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotEquals;
import static org.testng.Assert.assertTrue;
import io.netty.util.concurrent.FastThreadLocalThread;
import java.lang.reflect.Field;
Expand All @@ -33,7 +33,9 @@
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiFunction;
import lombok.AllArgsConstructor;
import lombok.Data;
Expand Down Expand Up @@ -104,7 +106,7 @@ private ProducerImpl overrideProducerForReplicator(AbstractReplicator replicator
return originalValue;
}

@Test
@Test(timeOut = 45 * 1000)
public void testReplicatorProducerStatInTopic() throws Exception {
final String topicName = BrokerTestUtil.newUniqueName("persistent://" + defaultNamespace + "/tp_");
final String subscribeName = "subscribe_1";
Expand All @@ -130,7 +132,7 @@ public void testReplicatorProducerStatInTopic() throws Exception {
});
}

@Test
@Test(timeOut = 45 * 1000)
public void testCreateRemoteConsumerFirst() throws Exception {
final String topicName = BrokerTestUtil.newUniqueName("persistent://" + defaultNamespace + "/tp_");
Producer<String> producer1 = client1.newProducer(Schema.STRING).topic(topicName).create();
Expand All @@ -150,28 +152,49 @@ public void testCreateRemoteConsumerFirst() throws Exception {
});
}

@Test
@Test(timeOut = 45 * 1000)
public void testTopicCloseWhenInternalProducerCloseErrorOnce() throws Exception {
final String topicName = BrokerTestUtil.newUniqueName("persistent://" + defaultNamespace + "/tp_");
admin1.topics().createNonPartitionedTopic(topicName);
// Wait for replicator started.
waitReplicatorStarted(topicName);
PersistentTopic persistentTopic =
PersistentTopic topic1 =
(PersistentTopic) pulsar1.getBrokerService().getTopic(topicName, false).join().get();
PersistentReplicator replicator =
(PersistentReplicator) persistentTopic.getReplicators().values().iterator().next();
PersistentReplicator replicator1 =
(PersistentReplicator) topic1.getReplicators().values().iterator().next();
// Mock an error when calling "replicator.disconnect()"
ProducerImpl mockProducer = Mockito.mock(ProducerImpl.class);
when(mockProducer.closeAsync()).thenReturn(CompletableFuture.failedFuture(new Exception("mocked ex")));
ProducerImpl originalProducer = overrideProducerForReplicator(replicator, mockProducer);
AtomicBoolean closeFailed = new AtomicBoolean(true);
final ProducerImpl mockProducer = Mockito.mock(ProducerImpl.class);
final AtomicReference<ProducerImpl> originalProducer1 = new AtomicReference();
doAnswer(invocation -> {
if (closeFailed.get()) {
return CompletableFuture.failedFuture(new Exception("mocked ex"));
} else {
return originalProducer1.get().closeAsync();
}
}).when(mockProducer).closeAsync();
originalProducer1.set(overrideProducerForReplicator(replicator1, mockProducer));
// Verify: since the "replicator.producer.closeAsync()" will retry after it failed, the topic unload should be
// successful.
admin1.topics().unload(topicName);
// Verify: After "replicator.producer.closeAsync()" retry again, the "replicator.producer" will be closed
// successful.
overrideProducerForReplicator(replicator, originalProducer);
closeFailed.set(false);
AtomicReference<PersistentTopic> topic2 = new AtomicReference();
AtomicReference<PersistentReplicator> replicator2 = new AtomicReference();
Awaitility.await().untilAsserted(() -> {
Assert.assertFalse(replicator.isConnected());
topic2.set((PersistentTopic) pulsar1.getBrokerService().getTopic(topicName, false).join().get());
replicator2.set((PersistentReplicator) topic2.get().getReplicators().values().iterator().next());
// It is a new Topic after reloading.
assertNotEquals(topic2.get(), topic1);
assertNotEquals(replicator2.get(), replicator1);
});
Awaitility.await().untilAsserted(() -> {
// Old replicator should be closed.
Assert.assertFalse(replicator1.isConnected());
Assert.assertFalse(originalProducer1.get().isConnected());
// New replicator should be connected.
Assert.assertTrue(replicator2.get().isConnected());
});
// cleanup.
cleanupTopics(() -> {
Expand Down Expand Up @@ -205,19 +228,26 @@ private void injectMockReplicatorProducerBuilder(
// Inject producer decorator.
doAnswer(invocation -> {
Schema schema = (Schema) invocation.getArguments()[0];
ProducerBuilderImpl producerBuilder = (ProducerBuilderImpl) internalClient.newProducer(schema);
ProducerBuilderImpl<?> producerBuilder = (ProducerBuilderImpl) internalClient.newProducer(schema);
ProducerBuilder spyProducerBuilder = spy(producerBuilder);
doAnswer(ignore -> {
CompletableFuture<Producer> producerFuture = new CompletableFuture<>();
final ProducerImpl p = (ProducerImpl) producerBuilder.create();
new FastThreadLocalThread(() -> {
try {
ProducerImpl newProducer = producerDecorator.apply(producerBuilder.getConf(), p);
producerFuture.complete(newProducer);
} catch (Exception ex) {
producerFuture.completeExceptionally(ex);
producerBuilder.createAsync().whenComplete((p, t) -> {
if (t != null) {
producerFuture.completeExceptionally(t);
return;
}
}).start();
ProducerImpl pImpl = (ProducerImpl) p;
new FastThreadLocalThread(() -> {
try {
ProducerImpl newProducer = producerDecorator.apply(producerBuilder.getConf(), pImpl);
producerFuture.complete(newProducer);
} catch (Exception ex) {
producerFuture.completeExceptionally(ex);
}
}).start();
});

return producerFuture;
}).when(spyProducerBuilder).createAsync();
return spyProducerBuilder;
Expand Down Expand Up @@ -306,7 +336,7 @@ void startCallback() {
* - The topic is wholly closed.
* - Verify: the delayed created internal producer will be closed.
*/
@Test
@Test(timeOut = 120 * 1000)
public void testConcurrencyOfUnloadBundleAndRecreateProducer() throws Exception {
final String topicName = BrokerTestUtil.newUniqueName("persistent://" + defaultNamespace + "/tp_");
// Inject an error for "replicator.producer" creation.
Expand Down Expand Up @@ -347,7 +377,8 @@ public void testConcurrencyOfUnloadBundleAndRecreateProducer() throws Exception
// Stuck start new producer, until the state of replicator change to Stopped.
// The next once of "createProducerSuccessAfterFailTimes" to create producer will be successfully.
Awaitility.await().pollInterval(Duration.ofMillis(100)).atMost(Duration.ofSeconds(60)).untilAsserted(() -> {
assertTrue(createProducerCounter.get() >= failTimes);
assertTrue(createProducerCounter.get() >= failTimes,
"count of retry to create producer is " + createProducerCounter.get());
});
CompletableFuture<Void> topicCloseFuture = persistentTopic.close(true);
Awaitility.await().atMost(Duration.ofSeconds(30)).untilAsserted(() -> {
Expand Down
38 changes: 38 additions & 0 deletions pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTestBase.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,21 +18,28 @@
*/
package org.apache.pulsar.broker.service;

import static org.apache.pulsar.compaction.Compactor.COMPACTION_SUBSCRIPTION;
import com.google.common.collect.Sets;
import java.net.URL;
import java.time.Duration;
import java.util.Collections;
import java.util.Optional;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.common.naming.SystemTopicNames;
import org.apache.pulsar.common.policies.data.ClusterData;
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
import org.apache.pulsar.common.policies.data.TopicType;
import org.apache.pulsar.common.policies.data.stats.TopicStatsImpl;
import org.apache.pulsar.tests.TestRetrySupport;
import org.apache.pulsar.zookeeper.LocalBookkeeperEnsemble;
import org.apache.pulsar.zookeeper.ZookeeperServerTest;
import org.awaitility.Awaitility;
import org.testng.Assert;

@Slf4j
public abstract class OneWayReplicatorTestBase extends TestRetrySupport {
Expand Down Expand Up @@ -140,10 +147,32 @@ protected void createDefaultTenantsAndClustersAndNamespace() throws Exception {
}

protected void cleanupTopics(CleanupTopicAction cleanupTopicAction) throws Exception {
waitChangeEventsInit(defaultNamespace);
admin1.namespaces().setNamespaceReplicationClusters(defaultNamespace, Collections.singleton(cluster1));
admin1.namespaces().unload(defaultNamespace);
cleanupTopicAction.run();
admin1.namespaces().setNamespaceReplicationClusters(defaultNamespace, Sets.newHashSet(cluster1, cluster2));
waitChangeEventsInit(defaultNamespace);
}

protected void waitChangeEventsInit(String namespace) {
PersistentTopic topic = (PersistentTopic) pulsar1.getBrokerService()
.getTopic(namespace + "/" + SystemTopicNames.NAMESPACE_EVENTS_LOCAL_NAME, false)
.join().get();
Awaitility.await().atMost(Duration.ofSeconds(180)).untilAsserted(() -> {
TopicStatsImpl topicStats = topic.getStats(true, false, false);
topicStats.getSubscriptions().entrySet().forEach(entry -> {
// No wait for compaction.
if (COMPACTION_SUBSCRIPTION.equals(entry.getKey())) {
return;
}
// No wait for durable cursor.
if (entry.getValue().isDurable()) {
return;
}
Assert.assertTrue(entry.getValue().getMsgBacklog() == 0, entry.getKey());
});
});
}

protected interface CleanupTopicAction {
Expand Down Expand Up @@ -185,10 +214,19 @@ protected void setConfigDefaults(ServiceConfiguration config, String clusterName
config.setAllowAutoTopicCreationType(TopicType.NON_PARTITIONED);
config.setEnableReplicatedSubscriptions(true);
config.setReplicatedSubscriptionsSnapshotFrequencyMillis(1000);
config.setLoadBalancerSheddingEnabled(false);
}

@Override
protected void cleanup() throws Exception {
// delete namespaces.
waitChangeEventsInit(defaultNamespace);
admin1.namespaces().setNamespaceReplicationClusters(defaultNamespace, Sets.newHashSet(cluster1));
admin1.namespaces().deleteNamespace(defaultNamespace);
admin2.namespaces().setNamespaceReplicationClusters(defaultNamespace, Sets.newHashSet(cluster2));
admin2.namespaces().deleteNamespace(defaultNamespace);

// shutdown.
markCurrentSetupNumberCleaned();
log.info("--- Shutting down ---");

Expand Down

0 comments on commit 5793ca1

Please sign in to comment.