| /** |
| * 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.service; |
| |
| import static org.apache.pulsar.broker.BrokerTestUtil.newUniqueName; |
| import static org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest.retryStrategically; |
| import static org.mockito.ArgumentMatchers.eq; |
| import static org.mockito.Mockito.doAnswer; |
| import static org.mockito.Mockito.spy; |
| import static org.testng.Assert.assertEquals; |
| import static org.testng.Assert.assertNotEquals; |
| import static org.testng.Assert.assertNotNull; |
| import static org.testng.Assert.assertNull; |
| import static org.testng.Assert.assertTrue; |
| import static org.testng.Assert.fail; |
| import com.google.common.collect.Sets; |
| import com.scurrilous.circe.checksum.Crc32cIntChecksum; |
| import io.netty.buffer.ByteBuf; |
| import java.lang.reflect.Field; |
| import java.lang.reflect.Method; |
| import java.nio.charset.StandardCharsets; |
| import java.util.ArrayList; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Optional; |
| import java.util.Set; |
| import java.util.SortedSet; |
| import java.util.TreeSet; |
| import java.util.UUID; |
| import java.util.concurrent.Callable; |
| import java.util.concurrent.CompletableFuture; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.CountDownLatch; |
| import java.util.concurrent.ExecutorService; |
| import java.util.concurrent.Executors; |
| import java.util.concurrent.Future; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.atomic.AtomicBoolean; |
| import java.util.stream.Collectors; |
| import lombok.Cleanup; |
| import org.apache.bookkeeper.mledger.AsyncCallbacks; |
| import org.apache.bookkeeper.mledger.AsyncCallbacks.DeleteCursorCallback; |
| import org.apache.bookkeeper.mledger.Entry; |
| import org.apache.bookkeeper.mledger.ManagedCursor; |
| import org.apache.bookkeeper.mledger.ManagedLedgerException; |
| import org.apache.bookkeeper.mledger.ManagedLedgerException.CursorAlreadyClosedException; |
| import org.apache.bookkeeper.mledger.Position; |
| import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl; |
| import org.apache.bookkeeper.mledger.impl.ManagedLedgerFactoryImpl; |
| import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl; |
| import org.apache.bookkeeper.mledger.impl.PositionImpl; |
| import org.apache.pulsar.broker.BrokerTestUtil; |
| import org.apache.pulsar.broker.PulsarService; |
| import org.apache.pulsar.broker.service.BrokerServiceException.NamingException; |
| import org.apache.pulsar.broker.service.persistent.PersistentReplicator; |
| import org.apache.pulsar.broker.service.persistent.PersistentTopic; |
| import org.apache.pulsar.client.admin.PulsarAdmin; |
| import org.apache.pulsar.client.admin.PulsarAdminException; |
| import org.apache.pulsar.client.api.Consumer; |
| import org.apache.pulsar.client.api.Message; |
| import org.apache.pulsar.client.api.MessageId; |
| import org.apache.pulsar.client.api.MessageRoutingMode; |
| import org.apache.pulsar.client.api.Producer; |
| import org.apache.pulsar.client.api.PulsarClient; |
| import org.apache.pulsar.client.api.PulsarClientException; |
| import org.apache.pulsar.client.api.RawMessage; |
| import org.apache.pulsar.client.api.RawReader; |
| import org.apache.pulsar.client.api.Schema; |
| import org.apache.pulsar.client.api.SubscriptionType; |
| import org.apache.pulsar.client.api.TypedMessageBuilder; |
| import org.apache.pulsar.client.api.schema.GenericRecord; |
| import org.apache.pulsar.client.api.schema.SchemaDefinition; |
| import org.apache.pulsar.client.impl.MessageIdImpl; |
| import org.apache.pulsar.client.impl.ProducerImpl; |
| import org.apache.pulsar.client.impl.PulsarClientImpl; |
| import org.apache.pulsar.client.impl.conf.ProducerConfigurationData; |
| import org.apache.pulsar.client.impl.transaction.TransactionImpl; |
| import org.apache.pulsar.common.events.EventsTopicNames; |
| import org.apache.pulsar.common.naming.NamespaceName; |
| import org.apache.pulsar.common.naming.TopicName; |
| import org.apache.pulsar.common.policies.data.BacklogQuota; |
| import org.apache.pulsar.common.policies.data.BacklogQuota.RetentionPolicy; |
| import org.apache.pulsar.common.policies.data.ClusterData; |
| import org.apache.pulsar.common.policies.data.PartitionedTopicStats; |
| import org.apache.pulsar.common.policies.data.ReplicatorStats; |
| import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy; |
| import org.apache.pulsar.common.policies.data.TopicStats; |
| import org.apache.pulsar.common.protocol.Commands; |
| import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap; |
| import org.apache.pulsar.schema.Schemas; |
| import org.awaitility.Awaitility; |
| import org.awaitility.reflect.WhiteboxImpl; |
| import org.mockito.Mockito; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| import org.testng.Assert; |
| import org.testng.annotations.AfterClass; |
| import org.testng.annotations.BeforeClass; |
| import org.testng.annotations.BeforeMethod; |
| import org.testng.annotations.DataProvider; |
| import org.testng.annotations.Test; |
| import org.testng.collections.Lists; |
| |
| /** |
| * Starts 3 brokers that are in 3 different clusters |
| */ |
| @Test(groups = "broker") |
| public class ReplicatorTest extends ReplicatorTestBase { |
| |
| protected String methodName; |
| |
| @BeforeMethod(alwaysRun = true) |
| public void beforeMethod(Method m) throws Exception { |
| methodName = m.getName(); |
| admin1.namespaces().removeBacklogQuota("pulsar/ns"); |
| admin1.namespaces().removeBacklogQuota("pulsar/ns1"); |
| admin1.namespaces().removeBacklogQuota("pulsar/global/ns"); |
| } |
| |
| @Override |
| @BeforeClass(alwaysRun = true, timeOut = 300000) |
| public void setup() throws Exception { |
| super.setup(); |
| } |
| |
| @Override |
| @AfterClass(alwaysRun = true, timeOut = 300000) |
| public void cleanup() throws Exception { |
| super.cleanup(); |
| } |
| |
| @DataProvider(name = "partitionedTopic") |
| public Object[][] partitionedTopicProvider() { |
| return new Object[][] { { Boolean.TRUE }, { Boolean.FALSE } }; |
| } |
| |
| @Test |
| public void testConfigChange() throws Exception { |
| log.info("--- Starting ReplicatorTest::testConfigChange ---"); |
| // This test is to verify that the config change on global namespace is successfully applied in broker during |
| // runtime. |
| // Run a set of producer tasks to create the topics |
| List<Future<Void>> results = Lists.newArrayList(); |
| for (int i = 0; i < 10; i++) { |
| final TopicName dest = TopicName.get(BrokerTestUtil.newUniqueName("persistent://pulsar/ns/topic-" + i)); |
| |
| results.add(executor.submit(new Callable<Void>() { |
| @Override |
| public Void call() throws Exception { |
| |
| @Cleanup |
| MessageProducer producer = new MessageProducer(url1, dest); |
| log.info("--- Starting producer --- " + url1); |
| |
| @Cleanup |
| MessageConsumer consumer = new MessageConsumer(url1, dest); |
| log.info("--- Starting Consumer --- " + url1); |
| |
| producer.produce(2); |
| consumer.receive(2); |
| return null; |
| } |
| })); |
| } |
| |
| for (Future<Void> result : results) { |
| try { |
| result.get(); |
| } catch (Exception e) { |
| log.error("exception in getting future result ", e); |
| fail(String.format("replication test failed with %s exception", e.getMessage())); |
| } |
| } |
| |
| Thread.sleep(1000L); |
| // Make sure that the internal replicators map contains remote cluster info |
| ConcurrentOpenHashMap<String, PulsarClient> replicationClients1 = ns1.getReplicationClients(); |
| ConcurrentOpenHashMap<String, PulsarClient> replicationClients2 = ns2.getReplicationClients(); |
| ConcurrentOpenHashMap<String, PulsarClient> replicationClients3 = ns3.getReplicationClients(); |
| |
| Assert.assertNotNull(replicationClients1.get("r2")); |
| Assert.assertNotNull(replicationClients1.get("r3")); |
| Assert.assertNotNull(replicationClients2.get("r1")); |
| Assert.assertNotNull(replicationClients2.get("r3")); |
| Assert.assertNotNull(replicationClients3.get("r1")); |
| Assert.assertNotNull(replicationClients3.get("r2")); |
| |
| // Case 1: Update the global namespace replication configuration to only contains the local cluster itself |
| admin1.namespaces().setNamespaceReplicationClusters("pulsar/ns", Sets.newHashSet("r1")); |
| |
| // Wait for config changes to be updated. |
| Thread.sleep(1000L); |
| |
| // Make sure that the internal replicators map still contains remote cluster info |
| Assert.assertNotNull(replicationClients1.get("r2")); |
| Assert.assertNotNull(replicationClients1.get("r3")); |
| Assert.assertNotNull(replicationClients2.get("r1")); |
| Assert.assertNotNull(replicationClients2.get("r3")); |
| Assert.assertNotNull(replicationClients3.get("r1")); |
| Assert.assertNotNull(replicationClients3.get("r2")); |
| |
| // Case 2: Update the configuration back |
| admin1.namespaces().setNamespaceReplicationClusters("pulsar/ns", Sets.newHashSet("r1", "r2", "r3")); |
| |
| // Wait for config changes to be updated. |
| Thread.sleep(1000L); |
| |
| // Make sure that the internal replicators map still contains remote cluster info |
| Assert.assertNotNull(replicationClients1.get("r2")); |
| Assert.assertNotNull(replicationClients1.get("r3")); |
| Assert.assertNotNull(replicationClients2.get("r1")); |
| Assert.assertNotNull(replicationClients2.get("r3")); |
| Assert.assertNotNull(replicationClients3.get("r1")); |
| Assert.assertNotNull(replicationClients3.get("r2")); |
| |
| // Case 3: TODO: Once automatic cleanup is implemented, add tests case to verify auto removal of clusters |
| } |
| |
| @Test(timeOut = 10000) |
| public void activeBrokerParse() throws Exception { |
| pulsar1.getConfiguration().setAuthorizationEnabled(true); |
| //init clusterData |
| |
| String cluster2ServiceUrls = String.format("%s,localhost:1234,localhost:5678,localhost:5677,localhost:5676", |
| pulsar2.getWebServiceAddress()); |
| ClusterData cluster2Data = ClusterData.builder().serviceUrl(cluster2ServiceUrls).build(); |
| String cluster2 = "activeCLuster2"; |
| admin2.clusters().createCluster(cluster2, cluster2Data); |
| Awaitility.await().until(() |
| -> admin2.clusters().getCluster(cluster2) != null); |
| |
| List<String> list = admin1.brokers().getActiveBrokers(cluster2); |
| assertEquals(list.get(0), url2.toString().replace("http://", "")); |
| //restore configuration |
| pulsar1.getConfiguration().setAuthorizationEnabled(false); |
| } |
| |
| @SuppressWarnings("unchecked") |
| @Test(timeOut = 30000) |
| public void testConcurrentReplicator() throws Exception { |
| |
| log.info("--- Starting ReplicatorTest::testConcurrentReplicator ---"); |
| |
| final String namespace = BrokerTestUtil.newUniqueName("pulsar/concurrent"); |
| admin1.namespaces().createNamespace(namespace); |
| admin1.namespaces().setNamespaceReplicationClusters(namespace, Sets.newHashSet("r1", "r2")); |
| final TopicName topicName = TopicName |
| .get(BrokerTestUtil.newUniqueName("persistent://" + namespace + "/topic")); |
| |
| @Cleanup |
| PulsarClient client1 = PulsarClient.builder() |
| .serviceUrl(url1.toString()).statsInterval(0, TimeUnit.SECONDS) |
| .build(); |
| Producer<byte[]> producer = client1.newProducer().topic(topicName.toString()) |
| .enableBatching(false) |
| .messageRoutingMode(MessageRoutingMode.SinglePartition) |
| .create(); |
| producer.close(); |
| |
| PersistentTopic topic = (PersistentTopic) pulsar1.getBrokerService() |
| .getOrCreateTopic(topicName.toString()).get(); |
| |
| PulsarClientImpl pulsarClient = spy((PulsarClientImpl) pulsar1.getBrokerService() |
| .getReplicationClient("r3", |
| pulsar1.getBrokerService().pulsar().getPulsarResources().getClusterResources() |
| .getCluster("r3"))); |
| final Method startRepl = PersistentTopic.class.getDeclaredMethod("startReplicator", String.class); |
| startRepl.setAccessible(true); |
| |
| Field replClientField = BrokerService.class.getDeclaredField("replicationClients"); |
| replClientField.setAccessible(true); |
| ConcurrentOpenHashMap<String, PulsarClient> replicationClients = |
| (ConcurrentOpenHashMap<String, PulsarClient>) replClientField |
| .get(pulsar1.getBrokerService()); |
| replicationClients.put("r3", pulsarClient); |
| |
| admin1.namespaces().setNamespaceReplicationClusters(namespace, Sets.newHashSet("r1", "r2", "r3")); |
| @Cleanup("shutdownNow") |
| ExecutorService executor = Executors.newFixedThreadPool(5); |
| for (int i = 0; i < 5; i++) { |
| executor.submit(() -> { |
| try { |
| startRepl.invoke(topic, "r3"); |
| } catch (Exception e) { |
| fail("setting replicator failed", e); |
| } |
| }); |
| } |
| Thread.sleep(3000); |
| // One time is to create a Replicator for user topics, |
| // and the other time is to create a Replicator for Topic Policies |
| Mockito.verify(pulsarClient, Mockito.times(2)) |
| .createProducerAsync( |
| Mockito.any(ProducerConfigurationData.class), |
| Mockito.any(Schema.class), eq(null)); |
| } |
| |
| @DataProvider(name = "namespace") |
| public Object[][] namespaceNameProvider() { |
| return new Object[][] { { "pulsar/ns" }, { "pulsar/global/ns" } }; |
| } |
| |
| @Test(dataProvider = "namespace") |
| public void testReplication(String namespace) throws Exception { |
| log.info("--- Starting ReplicatorTest::testReplication ---"); |
| |
| // This test is to verify that the config change on global namespace is successfully applied in broker during |
| // runtime. |
| // Run a set of producer tasks to create the topics |
| final TopicName dest = TopicName |
| .get(BrokerTestUtil.newUniqueName("persistent://" + namespace + "/repltopic")); |
| |
| @Cleanup |
| MessageProducer producer1 = new MessageProducer(url1, dest); |
| log.info("--- Starting producer --- " + url1); |
| |
| @Cleanup |
| MessageProducer producer2 = new MessageProducer(url2, dest); |
| log.info("--- Starting producer --- " + url2); |
| |
| @Cleanup |
| MessageProducer producer3 = new MessageProducer(url3, dest); |
| log.info("--- Starting producer --- " + url3); |
| |
| @Cleanup |
| MessageConsumer consumer1 = new MessageConsumer(url1, dest); |
| log.info("--- Starting Consumer --- " + url1); |
| |
| @Cleanup |
| MessageConsumer consumer2 = new MessageConsumer(url2, dest); |
| log.info("--- Starting Consumer --- " + url2); |
| |
| @Cleanup |
| MessageConsumer consumer3 = new MessageConsumer(url3, dest); |
| log.info("--- Starting Consumer --- " + url3); |
| |
| // Produce from cluster1 and consume from the rest |
| producer1.produce(2); |
| |
| consumer1.receive(2); |
| |
| consumer2.receive(2); |
| |
| consumer3.receive(2); |
| |
| // Produce from cluster2 and consume from the rest |
| producer2.produce(2); |
| |
| consumer1.receive(2); |
| |
| consumer2.receive(2); |
| |
| consumer3.receive(2); |
| |
| // Produce from cluster3 and consume from the rest |
| producer3.produce(2); |
| |
| consumer1.receive(2); |
| |
| consumer2.receive(2); |
| |
| consumer3.receive(2); |
| |
| // Produce from cluster1&2 and consume from cluster3 |
| producer1.produce(1); |
| producer2.produce(1); |
| |
| consumer1.receive(1); |
| |
| consumer2.receive(1); |
| |
| consumer3.receive(1); |
| |
| consumer1.receive(1); |
| |
| consumer2.receive(1); |
| |
| consumer3.receive(1); |
| } |
| |
| @Test(invocationCount = 5) |
| public void testReplicationWithSchema() throws Exception { |
| config1.setBrokerDeduplicationEnabled(true); |
| config2.setBrokerDeduplicationEnabled(true); |
| config3.setBrokerDeduplicationEnabled(true); |
| PulsarClient client1 = pulsar1.getClient(); |
| PulsarClient client2 = pulsar2.getClient(); |
| PulsarClient client3 = pulsar3.getClient(); |
| final TopicName topic = TopicName |
| .get(BrokerTestUtil.newUniqueName("persistent://pulsar/ns/testReplicationWithSchema")); |
| |
| final String subName = "my-sub"; |
| |
| @Cleanup |
| Producer<Schemas.PersonOne> producer1 = client1.newProducer(Schema.AVRO(Schemas.PersonOne.class)) |
| .topic(topic.toString()) |
| .enableBatching(false) |
| .create(); |
| |
| @Cleanup |
| Producer<Schemas.PersonThree> producer2 = client1.newProducer(Schema.AVRO(Schemas.PersonThree.class)) |
| .topic(topic.toString()) |
| .enableBatching(false) |
| .create(); |
| |
| admin1.topics().createSubscription(topic.toString(), subName, MessageId.earliest); |
| admin2.topics().createSubscription(topic.toString(), subName, MessageId.earliest); |
| admin3.topics().createSubscription(topic.toString(), subName, MessageId.earliest); |
| |
| final int totalMessages = 1000; |
| |
| for (int i = 0; i < totalMessages / 2; i++) { |
| producer1.sendAsync(new Schemas.PersonOne(i)); |
| } |
| |
| for (int i = 500; i < totalMessages; i++) { |
| producer2.sendAsync(new Schemas.PersonThree(i, "name-" + i)); |
| } |
| |
| Awaitility.await().untilAsserted(() -> { |
| assertTrue(admin1.topics().getInternalStats(topic.toString()).schemaLedgers.size() > 0); |
| assertTrue(admin2.topics().getInternalStats(topic.toString()).schemaLedgers.size() > 0); |
| assertTrue(admin3.topics().getInternalStats(topic.toString()).schemaLedgers.size() > 0); |
| }); |
| |
| @Cleanup |
| Consumer<GenericRecord> consumer1 = client1.newConsumer(Schema.AUTO_CONSUME()) |
| .topic(topic.toString()) |
| .subscriptionName(subName) |
| .subscribe(); |
| |
| @Cleanup |
| Consumer<GenericRecord> consumer2 = client2.newConsumer(Schema.AUTO_CONSUME()) |
| .topic(topic.toString()) |
| .subscriptionName(subName) |
| .subscribe(); |
| |
| @Cleanup |
| Consumer<GenericRecord> consumer3 = client3.newConsumer(Schema.AUTO_CONSUME()) |
| .topic(topic.toString()) |
| .subscriptionName(subName) |
| .subscribe(); |
| |
| int lastId = -1; |
| for (int i = 0; i < totalMessages; i++) { |
| Message<GenericRecord> msg1 = consumer1.receive(); |
| Message<GenericRecord> msg2 = consumer2.receive(); |
| Message<GenericRecord> msg3 = consumer3.receive(); |
| assertTrue(msg1 != null && msg2 != null && msg3 != null); |
| GenericRecord record1 = msg1.getValue(); |
| GenericRecord record2 = msg2.getValue(); |
| GenericRecord record3 = msg3.getValue(); |
| int id1 = (int) record1.getField("id"); |
| int id2 = (int) record2.getField("id"); |
| int id3 = (int) record3.getField("id"); |
| log.info("Received ids, id1: {}, id2: {}, id3: {}, lastId: {}", id1, id2, id3, lastId); |
| assertTrue(id1 == id2 && id2 == id3); |
| assertTrue(id1 > lastId); |
| lastId = id1; |
| consumer1.acknowledge(msg1); |
| consumer2.acknowledge(msg2); |
| consumer3.acknowledge(msg3); |
| } |
| |
| @Cleanup |
| Producer<byte[]> producerBytes = client1.newProducer() |
| .topic(topic.toString()) |
| .enableBatching(false) |
| .create(); |
| |
| byte[] data = "Bytes".getBytes(); |
| producerBytes.send(data); |
| |
| assertEquals(consumer1.receive().getValue().getNativeObject(), data); |
| assertEquals(consumer2.receive().getValue().getNativeObject(), data); |
| assertEquals(consumer3.receive().getValue().getNativeObject(), data); |
| } |
| |
| @Test |
| public void testCounterOfPendingMessagesCorrect() throws Exception { |
| // Init replicator and send many messages. |
| PulsarClient client1 = pulsar1.getClient(); |
| final TopicName topic = TopicName |
| .get(BrokerTestUtil.newUniqueName("persistent://pulsar/ns1/testReplicationWithGetSchemaError")); |
| final String subName = "my-sub"; |
| @Cleanup |
| Consumer<GenericRecord> consumer = client1.newConsumer(Schema.AUTO_CONSUME()) |
| .topic(topic.toString()) |
| .subscriptionName(subName) |
| .receiverQueueSize(10) |
| .subscribe(); |
| @Cleanup |
| Producer<Schemas.PersonOne> producer = client1.newProducer(Schema.AVRO(Schemas.PersonOne.class)) |
| .topic(topic.toString()) |
| .enableBatching(false) |
| .create(); |
| for (int i = 0; i < 20; i++) { |
| producer.send(new Schemas.PersonOne(i)); |
| } |
| |
| // Verify "pendingMessages" still is correct even if error occurs. |
| PersistentReplicator replicator = ensureReplicatorCreated(topic, pulsar1); |
| waitReplicateFinish(topic, admin1); |
| Awaitility.await().untilAsserted(() -> { |
| assertEquals((int) WhiteboxImpl.getInternalState(replicator, "pendingMessages"), 0); |
| }); |
| } |
| |
| @Test |
| public void testReplicationWillNotStuckByIncompleteSchemaFuture() throws Exception { |
| int originalReplicationProducerQueueSize = pulsar1.getConfiguration().getReplicationProducerQueueSize(); |
| pulsar1.getConfiguration().setReplicationProducerQueueSize(5); |
| // Init replicator and send many messages. |
| PulsarClient client1 = pulsar1.getClient(); |
| final TopicName topic = TopicName |
| .get(BrokerTestUtil.newUniqueName("persistent://pulsar/ns1/testReplicationWithGetSchemaError")); |
| admin1.namespaces().setSchemaCompatibilityStrategy("pulsar/ns1", SchemaCompatibilityStrategy.ALWAYS_COMPATIBLE); |
| final String subName = "sub"; |
| @Cleanup |
| Consumer<GenericRecord> consumer = client1.newConsumer(Schema.AUTO_CONSUME()) |
| .topic(topic.toString()) |
| .subscriptionName(subName) |
| .receiverQueueSize(10) |
| .subscribe(); |
| Thread sendTask = new Thread(() -> { |
| for (int i = 0; i < 50; i++) { |
| try { |
| Schema schema = Schema.JSON(SchemaDefinition.builder().withJsonDef(String.format( |
| "{" |
| + "\"type\": \"record\"," |
| + "\"name\": \"Test_Pojo\"," |
| + "\"namespace\": \"org.apache.pulsar.schema.compatibility\"," |
| + "\"fields\": [{" |
| + "\"name\": \"prop_%s\"," |
| + "\"type\": [\"null\", \"string\"]," |
| + "\"default\": null" |
| + "}]" |
| + "}", i)).build()); |
| Producer producer = client1 |
| .newProducer(schema) |
| .topic(topic.toString()) |
| .create(); |
| producer.send(String.valueOf(i).toString().getBytes()); |
| pulsar1.getBrokerService().checkReplicationPolicies(); |
| producer.close(); |
| Thread.sleep(100); |
| } catch (Exception e){ |
| } |
| } |
| }); |
| sendTask.start(); |
| |
| // Verify the replicate task can finish. |
| ensureReplicatorCreated(topic, pulsar1); |
| sendTask.join(); |
| waitReplicateFinish(topic, admin1); |
| |
| // cleanup |
| pulsar1.getConfiguration().setReplicationProducerQueueSize(originalReplicationProducerQueueSize); |
| } |
| |
| private static void waitReplicateFinish(TopicName topicName, PulsarAdmin admin){ |
| Awaitility.await().untilAsserted(() -> { |
| for (Map.Entry<String, ? extends ReplicatorStats> subStats : |
| admin.topics().getStats(topicName.toString(), true, false, false).getReplication().entrySet()){ |
| assertTrue(subStats.getValue().getReplicationBacklog() == 0, "replication task finished"); |
| } |
| }); |
| } |
| |
| private static PersistentReplicator ensureReplicatorCreated(TopicName topicName, PulsarService pulsar) { |
| PersistentTopic persistentTopic = |
| (PersistentTopic) pulsar.getBrokerService().getTopic(topicName.toString(), false).join().get(); |
| Awaitility.await().until(() -> !persistentTopic.getReplicators().isEmpty()); |
| return (PersistentReplicator) persistentTopic.getReplicators().values().iterator().next(); |
| } |
| |
| @Test |
| public void testReplicationOverrides() throws Exception { |
| log.info("--- Starting ReplicatorTest::testReplicationOverrides ---"); |
| |
| // This test is to verify that the config change on global namespace is successfully applied in broker during |
| // runtime. |
| // Run a set of producer tasks to create the topics |
| for (int i = 0; i < 10; i++) { |
| final TopicName dest = TopicName |
| .get(BrokerTestUtil.newUniqueName("persistent://pulsar/ns/repltopic")); |
| |
| @Cleanup |
| MessageProducer producer1 = new MessageProducer(url1, dest); |
| log.info("--- Starting producer --- " + url1); |
| |
| @Cleanup |
| MessageProducer producer2 = new MessageProducer(url2, dest); |
| log.info("--- Starting producer --- " + url2); |
| |
| @Cleanup |
| MessageProducer producer3 = new MessageProducer(url3, dest); |
| log.info("--- Starting producer --- " + url3); |
| |
| @Cleanup |
| MessageConsumer consumer1 = new MessageConsumer(url1, dest); |
| log.info("--- Starting Consumer --- " + url1); |
| |
| @Cleanup |
| MessageConsumer consumer2 = new MessageConsumer(url2, dest); |
| log.info("--- Starting Consumer --- " + url2); |
| |
| @Cleanup |
| MessageConsumer consumer3 = new MessageConsumer(url3, dest); |
| log.info("--- Starting Consumer --- " + url3); |
| |
| // Produce a message that isn't replicated |
| producer1.produce(1, producer1.newMessage().disableReplication()); |
| |
| consumer1.receive(1); |
| assertTrue(consumer2.drained()); |
| assertTrue(consumer3.drained()); |
| |
| // Produce a message not replicated to r2 |
| producer1.produce(1, producer1.newMessage().replicationClusters(Lists.newArrayList("r1", "r3"))); |
| consumer1.receive(1); |
| assertTrue(consumer2.drained()); |
| consumer3.receive(1); |
| |
| // Produce a default replicated message |
| producer1.produce(1); |
| |
| consumer1.receive(1); |
| consumer2.receive(1); |
| consumer3.receive(1); |
| |
| assertTrue(consumer1.drained()); |
| assertTrue(consumer2.drained()); |
| assertTrue(consumer3.drained()); |
| } |
| } |
| |
| @Test |
| public void testFailures() { |
| |
| log.info("--- Starting ReplicatorTest::testFailures ---"); |
| |
| try { |
| // 1. Create a consumer using the reserved consumer id prefix "pulsar.repl." |
| |
| final TopicName dest = TopicName |
| .get(BrokerTestUtil.newUniqueName("persistent://pulsar/ns/res-cons-id-")); |
| |
| // Create another consumer using replication prefix as sub id |
| MessageConsumer consumer = new MessageConsumer(url2, dest, "pulsar.repl."); |
| consumer.close(); |
| |
| } catch (Exception e) { |
| // SUCCESS |
| } |
| |
| } |
| |
| @Test(timeOut = 30000) |
| public void testReplicatePeekAndSkip() throws Exception { |
| |
| final TopicName dest = TopicName.get( |
| BrokerTestUtil.newUniqueName("persistent://pulsar/ns/peekAndSeekTopic")); |
| |
| @Cleanup |
| MessageProducer producer1 = new MessageProducer(url1, dest); |
| |
| @Cleanup |
| MessageConsumer consumer1 = new MessageConsumer(url3, dest); |
| |
| // Produce from cluster1 and consume from the rest |
| producer1.produce(2); |
| PersistentTopic topic = (PersistentTopic) pulsar1.getBrokerService().getTopicReference(dest.toString()).get(); |
| PersistentReplicator replicator = (PersistentReplicator) topic.getReplicators() |
| .get(topic.getReplicators().keys().get(0)); |
| replicator.skipMessages(2); |
| CompletableFuture<Entry> result = replicator.peekNthMessage(1); |
| Entry entry = result.get(50, TimeUnit.MILLISECONDS); |
| assertNull(entry); |
| } |
| |
| @Test(timeOut = 30000) |
| public void testReplicatorClearBacklog() throws Exception { |
| |
| // This test is to verify that reset cursor fails on global topic |
| SortedSet<String> testDests = new TreeSet<>(); |
| |
| final TopicName dest = TopicName |
| .get(BrokerTestUtil.newUniqueName("persistent://pulsar/ns/clearBacklogTopic")); |
| testDests.add(dest.toString()); |
| |
| @Cleanup |
| MessageProducer producer1 = new MessageProducer(url1, dest); |
| |
| @Cleanup |
| MessageConsumer consumer1 = new MessageConsumer(url3, dest); |
| |
| // Produce from cluster1 and consume from the rest |
| producer1.produce(2); |
| PersistentTopic topic = (PersistentTopic) pulsar1.getBrokerService().getTopicReference(dest.toString()).get(); |
| PersistentReplicator replicator = (PersistentReplicator) spy( |
| topic.getReplicators().get(topic.getReplicators().keys().get(0))); |
| replicator.readEntriesFailed(new ManagedLedgerException.InvalidCursorPositionException("failed"), null); |
| replicator.clearBacklog().get(); |
| Thread.sleep(100); |
| replicator.updateRates(); // for code-coverage |
| replicator.expireMessages(1); // for code-coverage |
| ReplicatorStats status = replicator.getStats(); |
| assertEquals(status.getReplicationBacklog(), 0); |
| } |
| |
| @Test(timeOut = 30000) |
| public void testResetCursorNotFail() throws Exception { |
| |
| log.info("--- Starting ReplicatorTest::testResetCursorNotFail ---"); |
| |
| // This test is to verify that reset cursor fails on global topic |
| final TopicName dest = TopicName |
| .get(BrokerTestUtil.newUniqueName("persistent://pulsar/ns/resetrepltopic")); |
| |
| @Cleanup |
| MessageProducer producer1 = new MessageProducer(url1, dest); |
| log.info("--- Starting producer --- " + url1); |
| |
| @Cleanup |
| MessageConsumer consumer1 = new MessageConsumer(url1, dest); |
| log.info("--- Starting Consumer --- " + url1); |
| |
| // Produce from cluster1 and consume from the rest |
| producer1.produce(2); |
| |
| consumer1.receive(2); |
| |
| admin1.topics().resetCursor(dest.toString(), "sub-id", System.currentTimeMillis()); |
| } |
| |
| @Test |
| public void testReplicationForBatchMessages() throws Exception { |
| log.info("--- Starting ReplicatorTest::testReplicationForBatchMessages ---"); |
| |
| // Run a set of producer tasks to create the topics |
| final TopicName dest = TopicName |
| .get(BrokerTestUtil.newUniqueName("persistent://pulsar/ns/repltopicbatch")); |
| |
| @Cleanup |
| MessageProducer producer1 = new MessageProducer(url1, dest, true); |
| log.info("--- Starting producer --- " + url1); |
| |
| @Cleanup |
| MessageProducer producer2 = new MessageProducer(url2, dest, true); |
| log.info("--- Starting producer --- " + url2); |
| |
| @Cleanup |
| MessageProducer producer3 = new MessageProducer(url3, dest, true); |
| log.info("--- Starting producer --- " + url3); |
| |
| @Cleanup |
| MessageConsumer consumer1 = new MessageConsumer(url1, dest); |
| log.info("--- Starting Consumer --- " + url1); |
| |
| @Cleanup |
| MessageConsumer consumer2 = new MessageConsumer(url2, dest); |
| log.info("--- Starting Consumer --- " + url2); |
| |
| @Cleanup |
| MessageConsumer consumer3 = new MessageConsumer(url3, dest); |
| log.info("--- Starting Consumer --- " + url3); |
| |
| // Produce from cluster1 and consume from the rest |
| producer1.produceBatch(10); |
| |
| consumer1.receive(10); |
| |
| consumer2.receive(10); |
| |
| consumer3.receive(10); |
| |
| // Produce from cluster2 and consume from the rest |
| producer2.produceBatch(10); |
| |
| consumer1.receive(10); |
| |
| consumer2.receive(10); |
| |
| consumer3.receive(10); |
| } |
| |
| /** |
| * It verifies that: if it fails while removing replicator-cluster-cursor: it should not restart the replicator and |
| * it should have cleaned up from the list |
| * |
| * @throws Exception |
| */ |
| @Test(timeOut = 30000) |
| public void testDeleteReplicatorFailure() throws Exception { |
| log.info("--- Starting ReplicatorTest::testDeleteReplicatorFailure ---"); |
| final String topicName = BrokerTestUtil.newUniqueName("persistent://pulsar/ns/repltopicbatch"); |
| final TopicName dest = TopicName.get(topicName); |
| |
| @Cleanup |
| MessageProducer producer1 = new MessageProducer(url1, dest); |
| PersistentTopic topic = (PersistentTopic) pulsar1.getBrokerService().getTopicReference(topicName).get(); |
| final String replicatorClusterName = topic.getReplicators().keys().get(0); |
| ManagedLedgerImpl ledger = (ManagedLedgerImpl) topic.getManagedLedger(); |
| CountDownLatch latch = new CountDownLatch(1); |
| // delete cursor already : so next time if topic.removeReplicator will get exception but then it should |
| // remove-replicator from the list even with failure |
| ledger.asyncDeleteCursor("pulsar.repl." + replicatorClusterName, new DeleteCursorCallback() { |
| @Override |
| public void deleteCursorComplete(Object ctx) { |
| latch.countDown(); |
| } |
| |
| @Override |
| public void deleteCursorFailed(ManagedLedgerException exception, Object ctx) { |
| latch.countDown(); |
| } |
| }, null); |
| latch.await(); |
| |
| Method removeReplicator = PersistentTopic.class.getDeclaredMethod("removeReplicator", String.class); |
| removeReplicator.setAccessible(true); |
| // invoke removeReplicator : it fails as cursor is not present: but still it should remove the replicator from |
| // list without restarting it |
| @SuppressWarnings("unchecked") |
| CompletableFuture<Void> result = (CompletableFuture<Void>) removeReplicator.invoke(topic, |
| replicatorClusterName); |
| result.thenApply((v) -> { |
| assertNull(topic.getPersistentReplicator(replicatorClusterName)); |
| return null; |
| }); |
| } |
| |
| @SuppressWarnings("unchecked") |
| @Test(priority = 5, timeOut = 30000) |
| public void testReplicatorProducerClosing() throws Exception { |
| log.info("--- Starting ReplicatorTest::testDeleteReplicatorFailure ---"); |
| final String topicName = BrokerTestUtil.newUniqueName("persistent://pulsar/ns/repltopicbatch"); |
| final TopicName dest = TopicName.get(topicName); |
| |
| @Cleanup |
| MessageProducer producer1 = new MessageProducer(url1, dest); |
| PersistentTopic topic = (PersistentTopic) pulsar1.getBrokerService().getTopicReference(topicName).get(); |
| final String replicatorClusterName = topic.getReplicators().keys().get(0); |
| Replicator replicator = topic.getPersistentReplicator(replicatorClusterName); |
| pulsar2.close(); |
| pulsar2 = null; |
| pulsar3.close(); |
| pulsar3 = null; |
| replicator.disconnect(false); |
| Thread.sleep(100); |
| Field field = AbstractReplicator.class.getDeclaredField("producer"); |
| field.setAccessible(true); |
| ProducerImpl<byte[]> producer = (ProducerImpl<byte[]>) field.get(replicator); |
| assertNull(producer); |
| } |
| |
| @Test(priority = 5, timeOut = 30000) |
| public void testReplicatorConnected() throws Exception { |
| final String topicName = BrokerTestUtil.newUniqueName("persistent://pulsar/ns/tp_" + UUID.randomUUID()); |
| final TopicName dest = TopicName.get(topicName); |
| admin1.topics().createPartitionedTopic(topicName, 1); |
| |
| @Cleanup |
| MessageProducer producer1 = new MessageProducer(url1, dest); |
| |
| Awaitility.await().until(() -> { |
| TopicStats topicStats = admin1.topics().getStats(topicName + "-partition-0"); |
| return topicStats.getReplication().values().stream() |
| .map(ReplicatorStats::isConnected).reduce((a, b) -> a & b).get(); |
| }); |
| |
| PartitionedTopicStats |
| partitionedTopicStats = admin1.topics().getPartitionedStats(topicName, true); |
| |
| for (ReplicatorStats replicatorStats : partitionedTopicStats.getReplication().values()){ |
| assertTrue(replicatorStats.isConnected()); |
| } |
| } |
| |
| @Test |
| public void testDeleteTopicFailure() throws Exception { |
| final String topicName = BrokerTestUtil.newUniqueName("persistent://pulsar/ns/tp_" + UUID.randomUUID()); |
| admin1.topics().createNonPartitionedTopic(topicName); |
| try { |
| admin1.topics().delete(topicName); |
| fail("Delete topic should fail if enabled replicator"); |
| } catch (Exception ex) { |
| assertTrue(ex instanceof PulsarAdminException); |
| assertEquals(((PulsarAdminException) ex).getStatusCode(), 422/* Unprocessable entity*/); |
| } |
| } |
| |
| @Test |
| public void testDeletePartitionedTopicFailure() throws Exception { |
| final String topicName = BrokerTestUtil.newUniqueName("persistent://pulsar/ns/tp_" + UUID.randomUUID()); |
| admin1.topics().createPartitionedTopic(topicName, 2); |
| admin1.topics().createSubscription(topicName, "sub1", MessageId.earliest); |
| try { |
| admin1.topics().deletePartitionedTopic(topicName); |
| fail("Delete topic should fail if enabled replicator"); |
| } catch (Exception ex) { |
| assertTrue(ex instanceof PulsarAdminException); |
| assertEquals(((PulsarAdminException) ex).getStatusCode(), 422/* Unprocessable entity*/); |
| } |
| } |
| |
| @Test(priority = 4, timeOut = 30000) |
| public void testReplicatorProducerName() throws Exception { |
| log.info("--- Starting ReplicatorTest::testReplicatorProducerName ---"); |
| final String topicName = BrokerTestUtil.newUniqueName("persistent://pulsar/ns/testReplicatorProducerName"); |
| final TopicName dest = TopicName.get(topicName); |
| |
| @Cleanup |
| MessageProducer producer1 = new MessageProducer(url1, dest); |
| |
| Awaitility.await().untilAsserted(() -> { |
| assertTrue(pulsar2.getBrokerService().getTopicReference(topicName).isPresent()); |
| }); |
| Optional<Topic> topic = pulsar2.getBrokerService().getTopicReference(topicName); |
| assertTrue(topic.isPresent()); |
| Awaitility.await().untilAsserted(() -> { |
| Set<String> remoteClusters = topic.get().getProducers().values().stream() |
| .map(org.apache.pulsar.broker.service.Producer::getRemoteCluster) |
| .collect(Collectors.toSet()); |
| assertTrue(remoteClusters.contains("r1")); |
| }); |
| } |
| |
| @Test(priority = 4, timeOut = 30000) |
| public void testReplicatorProducerNameWithUserDefinedReplicatorPrefix() throws Exception { |
| log.info("--- Starting ReplicatorTest::testReplicatorProducerNameWithUserDefinedReplicatorPrefix ---"); |
| final String topicName = BrokerTestUtil.newUniqueName( |
| "persistent://pulsar/ns/testReplicatorProducerNameWithUserDefinedReplicatorPrefix"); |
| final TopicName dest = TopicName.get(topicName); |
| |
| pulsar1.getConfiguration().setReplicatorPrefix("user-defined-prefix"); |
| pulsar2.getConfiguration().setReplicatorPrefix("user-defined-prefix"); |
| pulsar3.getConfiguration().setReplicatorPrefix("user-defined-prefix"); |
| |
| @Cleanup |
| MessageProducer producer1 = new MessageProducer(url1, dest); |
| |
| Awaitility.await().untilAsserted(()->{ |
| assertTrue(pulsar2.getBrokerService().getTopicReference(topicName).isPresent()); |
| }); |
| Optional<Topic> topic = pulsar2.getBrokerService().getTopicReference(topicName); |
| assertTrue(topic.isPresent()); |
| Set<String> remoteClusters = topic.get().getProducers().values().stream() |
| .map(org.apache.pulsar.broker.service.Producer::getRemoteCluster) |
| .collect(Collectors.toSet()); |
| assertTrue(remoteClusters.contains("r1")); |
| } |
| |
| |
| /** |
| * Issue #199 |
| * |
| * It verifies that: if the remote cluster reaches backlog quota limit, replicator temporarily stops and once the |
| * backlog drains it should resume replication. |
| * |
| * @throws Exception |
| */ |
| |
| @Test(timeOut = 60000, priority = -1) |
| public void testResumptionAfterBacklogRelaxed() throws Exception { |
| List<RetentionPolicy> policies = Lists.newArrayList(); |
| policies.add(RetentionPolicy.producer_exception); |
| policies.add(RetentionPolicy.producer_request_hold); |
| |
| for (RetentionPolicy policy : policies) { |
| // Use 1Mb quota by default |
| admin1.namespaces().setBacklogQuota("pulsar/ns1", BacklogQuota.builder() |
| .limitSize(1 * 1024 * 1024) |
| .retentionPolicy(policy) |
| .build()); |
| Thread.sleep(200); |
| |
| TopicName dest = TopicName |
| .get(BrokerTestUtil.newUniqueName("persistent://pulsar/ns1/%s-" + policy)); |
| |
| // Producer on r1 |
| @Cleanup |
| MessageProducer producer1 = new MessageProducer(url1, dest); |
| |
| // Consumer on r2 |
| @Cleanup |
| MessageConsumer consumer2 = new MessageConsumer(url2, dest); |
| |
| // Replicator for r1 -> r2 |
| PersistentTopic topic = (PersistentTopic) pulsar1.getBrokerService().getTopicReference(dest.toString()).get(); |
| Replicator replicator = topic.getPersistentReplicator("r2"); |
| |
| // Produce 1 message in r1. This message will be replicated immediately into r2 and it will become part of |
| // local backlog |
| producer1.produce(1); |
| |
| Thread.sleep(500); |
| |
| // Restrict backlog quota limit to 1 byte to stop replication |
| admin1.namespaces().setBacklogQuota("pulsar/ns1", BacklogQuota.builder() |
| .limitSize(1) |
| .retentionPolicy(policy) |
| .build()); |
| |
| Thread.sleep((TIME_TO_CHECK_BACKLOG_QUOTA + 1) * 1000); |
| |
| assertEquals(replicator.getStats().replicationBacklog, 0); |
| |
| // Next message will not be replicated, because r2 has reached the quota |
| producer1.produce(1); |
| |
| Thread.sleep(500); |
| |
| assertEquals(replicator.getStats().replicationBacklog, 1); |
| |
| // Consumer will now drain 1 message and the replication backlog will be cleared |
| consumer2.receive(1); |
| |
| // Wait until the 2nd message got delivered to consumer |
| consumer2.receive(1); |
| |
| int retry = 10; |
| for (int i = 0; i < retry && replicator.getStats().replicationBacklog > 0; i++) { |
| if (i != retry - 1) { |
| Thread.sleep(100); |
| } |
| } |
| |
| assertEquals(replicator.getStats().replicationBacklog, 0); |
| } |
| } |
| |
| /** |
| * It verifies that PersistentReplicator considers CursorAlreadyClosedException as non-retriable-read exception and |
| * it should closed the producer as cursor is already closed because replicator is already deleted. |
| * |
| * @throws Exception |
| */ |
| @Test(timeOut = 15000) |
| public void testCloseReplicatorStartProducer() throws Exception { |
| TopicName dest = TopicName.get(BrokerTestUtil.newUniqueName("persistent://pulsar/ns1/closeCursor")); |
| // Producer on r1 |
| @Cleanup |
| MessageProducer producer1 = new MessageProducer(url1, dest); |
| // Consumer on r1 |
| @Cleanup |
| MessageConsumer consumer1 = new MessageConsumer(url1, dest); |
| // Consumer on r2 |
| @Cleanup |
| MessageConsumer consumer2 = new MessageConsumer(url2, dest); |
| |
| // Replicator for r1 -> r2 |
| PersistentTopic topic = (PersistentTopic) pulsar1.getBrokerService().getTopicReference(dest.toString()).get(); |
| PersistentReplicator replicator = (PersistentReplicator) topic.getPersistentReplicator("r2"); |
| |
| // close the cursor |
| Field cursorField = PersistentReplicator.class.getDeclaredField("cursor"); |
| cursorField.setAccessible(true); |
| ManagedCursor cursor = (ManagedCursor) cursorField.get(replicator); |
| cursor.close(); |
| // try to read entries |
| producer1.produce(10); |
| |
| try { |
| cursor.readEntriesOrWait(10); |
| fail("It should have failed"); |
| } catch (Exception e) { |
| assertEquals(e.getClass(), CursorAlreadyClosedException.class); |
| } |
| |
| // replicator-readException: cursorAlreadyClosed |
| replicator.readEntriesFailed(new CursorAlreadyClosedException("Cursor already closed exception"), null); |
| |
| // wait replicator producer to be closed |
| Thread.sleep(100); |
| |
| // Replicator producer must be closed |
| Field producerField = AbstractReplicator.class.getDeclaredField("producer"); |
| producerField.setAccessible(true); |
| @SuppressWarnings("unchecked") |
| ProducerImpl<byte[]> replicatorProducer = (ProducerImpl<byte[]>) producerField.get(replicator); |
| assertNull(replicatorProducer); |
| } |
| |
| @Test(timeOut = 30000) |
| public void verifyChecksumAfterReplication() throws Exception { |
| final String topicName = BrokerTestUtil.newUniqueName("persistent://pulsar/ns/checksumAfterReplication"); |
| |
| @Cleanup |
| PulsarClient c1 = PulsarClient.builder().serviceUrl(url1.toString()).build(); |
| Producer<byte[]> p1 = c1.newProducer().topic(topicName) |
| .enableBatching(false) |
| .messageRoutingMode(MessageRoutingMode.SinglePartition) |
| .create(); |
| |
| @Cleanup |
| PulsarClient c2 = PulsarClient.builder().serviceUrl(url2.toString()).build(); |
| RawReader reader2 = RawReader.create(c2, topicName, "sub").get(); |
| |
| p1.send("Hello".getBytes()); |
| |
| RawMessage msg = reader2.readNextAsync().get(); |
| |
| ByteBuf b = msg.getHeadersAndPayload(); |
| |
| assertTrue(Commands.hasChecksum(b)); |
| int parsedChecksum = Commands.readChecksum(b); |
| int computedChecksum = Crc32cIntChecksum.computeChecksum(b); |
| |
| assertEquals(parsedChecksum, computedChecksum); |
| |
| p1.close(); |
| reader2.closeAsync().get(); |
| } |
| |
| @Test |
| public void testReplicatorWithPartitionedTopic() throws Exception { |
| final String namespace = "pulsar/partitionedNs-" + UUID.randomUUID(); |
| final String persistentTopicName = "persistent://" + namespace + "/partTopic" + UUID.randomUUID(); |
| |
| admin1.namespaces().createNamespace(namespace); |
| admin1.namespaces().setNamespaceReplicationClusters(namespace, Sets.newHashSet("r1", "r2", "r3")); |
| // Create partitioned-topic from R1 |
| admin1.topics().createPartitionedTopic(persistentTopicName, 3); |
| // List partitioned topics from R2 |
| Awaitility.await().untilAsserted(() -> assertNotNull(admin2.topics().getPartitionedTopicList(namespace))); |
| Awaitility.await().untilAsserted(() -> assertEquals( |
| admin2.topics().getPartitionedTopicList(namespace).get(0), persistentTopicName)); |
| assertEquals(admin1.topics().getList(namespace).size(), 3); |
| // List partitioned topics from R3 |
| Awaitility.await().untilAsserted(() -> assertNotNull(admin3.topics().getPartitionedTopicList(namespace))); |
| Awaitility.await().untilAsserted(() -> assertEquals( |
| admin3.topics().getPartitionedTopicList(namespace).get(0), persistentTopicName)); |
| // Update partitioned topic from R2 |
| admin2.topics().updatePartitionedTopic(persistentTopicName, 5); |
| assertEquals(admin2.topics().getPartitionedTopicMetadata(persistentTopicName).partitions, 5); |
| assertEquals((int) admin2.topics().getList(namespace).stream().filter(topic -> |
| !topic.contains(EventsTopicNames.NAMESPACE_EVENTS_LOCAL_NAME)).count(), 5); |
| // Update partitioned topic from R3 |
| admin3.topics().updatePartitionedTopic(persistentTopicName, 6); |
| assertEquals(admin3.topics().getPartitionedTopicMetadata(persistentTopicName).partitions, 6); |
| assertEquals(admin3.topics().getList(namespace).stream().filter(topic -> |
| !topic.contains(EventsTopicNames.NAMESPACE_EVENTS_LOCAL_NAME)).count(), 6); |
| // Update partitioned topic from R1 |
| admin1.topics().updatePartitionedTopic(persistentTopicName, 7); |
| assertEquals(admin1.topics().getPartitionedTopicMetadata(persistentTopicName).partitions, 7); |
| assertEquals(admin2.topics().getPartitionedTopicMetadata(persistentTopicName).partitions, 7); |
| assertEquals(admin3.topics().getPartitionedTopicMetadata(persistentTopicName).partitions, 7); |
| assertEquals(admin1.topics().getList(namespace).stream().filter(topic -> |
| !topic.contains(EventsTopicNames.NAMESPACE_EVENTS_LOCAL_NAME)).count(), 7); |
| assertEquals(admin2.topics().getList(namespace).stream().filter(topic -> |
| !topic.contains(EventsTopicNames.NAMESPACE_EVENTS_LOCAL_NAME)).count(), 7); |
| assertEquals(admin3.topics().getList(namespace).stream().filter(topic -> |
| !topic.contains(EventsTopicNames.NAMESPACE_EVENTS_LOCAL_NAME)).count(), 7); |
| } |
| |
| /** |
| * It verifies that broker should not start replicator for partitioned-topic (topic without -partition postfix) |
| * |
| * @param isPartitionedTopic |
| * @throws Exception |
| */ |
| @Test(dataProvider = "partitionedTopic") |
| public void testReplicatorOnPartitionedTopic(boolean isPartitionedTopic) throws Exception { |
| |
| log.info("--- Starting ReplicatorTest::{} --- ", methodName); |
| |
| final String namespace = BrokerTestUtil.newUniqueName("pulsar/partitionedNs-" + isPartitionedTopic); |
| final String persistentTopicName = BrokerTestUtil.newUniqueName("persistent://" + namespace + "/partTopic-" + isPartitionedTopic); |
| final String nonPersistentTopicName = BrokerTestUtil.newUniqueName("non-persistent://" + namespace + "/partTopic-" + isPartitionedTopic); |
| BrokerService brokerService = pulsar1.getBrokerService(); |
| |
| admin1.namespaces().createNamespace(namespace); |
| admin1.namespaces().setNamespaceReplicationClusters(namespace, Sets.newHashSet("r1", "r2", "r3")); |
| |
| if (isPartitionedTopic) { |
| admin1.topics().createPartitionedTopic(persistentTopicName, 5); |
| admin1.topics().createPartitionedTopic(nonPersistentTopicName, 5); |
| } |
| |
| // load namespace with dummy topic on ns |
| @Cleanup |
| PulsarClient client = PulsarClient.builder().serviceUrl(url1.toString()).build(); |
| client.newProducer().topic("persistent://" + namespace + "/dummyTopic") |
| .enableBatching(false) |
| .messageRoutingMode(MessageRoutingMode.SinglePartition) |
| .create(); |
| |
| // persistent topic test |
| try { |
| brokerService.getOrCreateTopic(persistentTopicName).get(); |
| if (isPartitionedTopic) { |
| fail("Topic creation fails with partitioned topic as replicator init fails"); |
| } |
| } catch (Exception e) { |
| if (!isPartitionedTopic) { |
| fail("Topic creation should not fail without any partitioned topic"); |
| } |
| assertTrue(e.getCause() instanceof NamingException); |
| } |
| |
| // non-persistent topic test |
| try { |
| brokerService.getOrCreateTopic(nonPersistentTopicName).get(); |
| if (isPartitionedTopic) { |
| fail("Topic creation fails with partitioned topic as replicator init fails"); |
| } |
| } catch (Exception e) { |
| if (!isPartitionedTopic) { |
| fail("Topic creation should not fail without any partitioned topic"); |
| } |
| assertTrue(e.getCause() instanceof NamingException); |
| } |
| |
| } |
| |
| @Test |
| public void testReplicatedCluster() throws Exception { |
| |
| log.info("--- Starting ReplicatorTest::testReplicatedCluster ---"); |
| |
| final String namespace = "pulsar/global/repl"; |
| final String topicName = BrokerTestUtil.newUniqueName("persistent://" + namespace + "/topic1"); |
| admin1.namespaces().createNamespace(namespace); |
| admin1.namespaces().setNamespaceReplicationClusters(namespace, Sets.newHashSet("r1", "r2", "r3")); |
| admin1.topics().createPartitionedTopic(topicName, 4); |
| |
| @Cleanup |
| PulsarClient client1 = PulsarClient.builder().serviceUrl(url1.toString()).statsInterval(0, TimeUnit.SECONDS) |
| .build(); |
| @Cleanup |
| PulsarClient client2 = PulsarClient.builder().serviceUrl(url2.toString()).statsInterval(0, TimeUnit.SECONDS) |
| .build(); |
| |
| Producer<byte[]> producer1 = client1.newProducer().topic(topicName).create(); |
| org.apache.pulsar.client.api.Consumer<byte[]> consumer1 = client1.newConsumer().topic(topicName).subscriptionName("s1").subscribe(); |
| org.apache.pulsar.client.api.Consumer<byte[]> consumer2 = client2.newConsumer().topic(topicName).subscriptionName("s1").subscribe(); |
| byte[] value = "test".getBytes(); |
| |
| // publish message local only |
| TypedMessageBuilder<byte[]> msg = producer1.newMessage().replicationClusters(Lists.newArrayList("r1")).value(value); |
| msg.send(); |
| assertEquals(consumer1.receive().getValue(), value); |
| |
| Message<byte[]> msg2 = consumer2.receive(1, TimeUnit.SECONDS); |
| if (msg2 != null) { |
| fail("msg should have not been replicated to remote cluster"); |
| } |
| |
| consumer1.close(); |
| consumer2.close(); |
| producer1.close(); |
| |
| } |
| |
| /** |
| * This validates that broker supports update-partition api for global topics. |
| * <pre> |
| * 1. Create global topic with 4 partitions |
| * 2. Update partition with 8 partitions |
| * 3. Create producer on the partition topic which loads all new partitions |
| * 4. Check subscriptions are created on all new partitions. |
| * </pre> |
| * @throws Exception |
| */ |
| @Test |
| public void testUpdateGlobalTopicPartition() throws Exception { |
| log.info("--- Starting ReplicatorTest::testUpdateGlobalTopicPartition ---"); |
| |
| final String cluster1 = pulsar1.getConfig().getClusterName(); |
| final String cluster2 = pulsar2.getConfig().getClusterName(); |
| final String namespace = BrokerTestUtil.newUniqueName("pulsar/ns"); |
| final String topicName = BrokerTestUtil.newUniqueName("persistent://" + namespace + "/topic1"); |
| int startPartitions = 4; |
| int newPartitions = 8; |
| final String subscriberName = "sub1"; |
| admin1.namespaces().createNamespace(namespace, Sets.newHashSet(cluster1, cluster2)); |
| admin1.topics().createPartitionedTopic(topicName, startPartitions); |
| |
| @Cleanup |
| PulsarClient client1 = PulsarClient.builder().serviceUrl(url1.toString()).statsInterval(0, TimeUnit.SECONDS) |
| .build(); |
| @Cleanup |
| PulsarClient client2 = PulsarClient.builder().serviceUrl(url2.toString()).statsInterval(0, TimeUnit.SECONDS) |
| .build(); |
| |
| Consumer<byte[]> consumer1 = client1.newConsumer().topic(topicName).subscriptionName(subscriberName) |
| .subscribe(); |
| Consumer<byte[]> consumer2 = client2.newConsumer().topic(topicName).subscriptionName(subscriberName) |
| .subscribe(); |
| |
| admin1.topics().updatePartitionedTopic(topicName, newPartitions); |
| |
| assertEquals(admin1.topics().getPartitionedTopicMetadata(topicName).partitions, newPartitions); |
| |
| // create producers to load all the partition topics |
| Producer<byte[]> producer1 = client1.newProducer().topic(topicName).create(); |
| Producer<byte[]> producer2 = client2.newProducer().topic(topicName).create(); |
| |
| for (int i = startPartitions; i < newPartitions; i++) { |
| String partitionedTopic = topicName + TopicName.PARTITIONED_TOPIC_SUFFIX + i; |
| assertEquals(admin1.topics().getSubscriptions(partitionedTopic).size(), 1); |
| assertEquals(admin2.topics().getSubscriptions(partitionedTopic).size(), 1); |
| } |
| |
| producer1.close(); |
| producer2.close(); |
| consumer1.close(); |
| consumer2.close(); |
| } |
| |
| @Test |
| public void testIncrementPartitionsOfTopicWithReplicatedSubscription() throws Exception { |
| final String cluster1 = pulsar1.getConfig().getClusterName(); |
| final String cluster2 = pulsar2.getConfig().getClusterName(); |
| final String namespace = BrokerTestUtil.newUniqueName("pulsar/ns"); |
| final String topicName = BrokerTestUtil.newUniqueName("persistent://" + namespace + "/topic1"); |
| int startPartitions = 4; |
| int newPartitions = 8; |
| final String subscriberName = "sub1"; |
| admin1.namespaces().createNamespace(namespace, Sets.newHashSet(cluster1, cluster2)); |
| admin1.topics().createPartitionedTopic(topicName, startPartitions); |
| |
| @Cleanup |
| PulsarClient client1 = PulsarClient.builder().serviceUrl(url1.toString()).statsInterval(0, TimeUnit.SECONDS) |
| .build(); |
| |
| Consumer<byte[]> consumer1 = client1.newConsumer().topic(topicName).subscriptionName(subscriberName) |
| .replicateSubscriptionState(true) |
| .subscribe(); |
| |
| admin1.topics().updatePartitionedTopic(topicName, newPartitions); |
| |
| assertEquals(admin1.topics().getPartitionedTopicMetadata(topicName).partitions, newPartitions); |
| |
| Map<String, Boolean> replicatedSubscriptionStatus = |
| admin1.topics().getReplicatedSubscriptionStatus(topicName, subscriberName); |
| assertEquals(replicatedSubscriptionStatus.size(), newPartitions); |
| for (Map.Entry<String, Boolean> replicatedStatusForPartition : replicatedSubscriptionStatus.entrySet()) { |
| assertTrue(replicatedStatusForPartition.getValue(), |
| "Replicated status is invalid for " + replicatedStatusForPartition.getKey()); |
| } |
| consumer1.close(); |
| } |
| |
| @DataProvider(name = "topicPrefix") |
| public static Object[][] topicPrefix() { |
| return new Object[][] { { "persistent://", "/persistent" }, { "non-persistent://", "/non-persistent" } }; |
| } |
| |
| @Test(dataProvider = "topicPrefix") |
| public void testTopicReplicatedAndProducerCreate(String topicPrefix, String topicName) throws Exception { |
| log.info("--- Starting ReplicatorTest::testTopicReplicatedAndProducerCreate ---"); |
| |
| final String cluster1 = pulsar1.getConfig().getClusterName(); |
| final String cluster2 = pulsar2.getConfig().getClusterName(); |
| final String namespace = BrokerTestUtil.newUniqueName("pulsar/ns"); |
| final String partitionedTopicName = BrokerTestUtil.newUniqueName(topicPrefix + namespace + topicName + "-partitioned"); |
| final String nonPartitionedTopicName = BrokerTestUtil.newUniqueName(topicPrefix + namespace + topicName + "-non-partitioned"); |
| final int startPartitions = 4; |
| admin1.namespaces().createNamespace(namespace, Sets.newHashSet(cluster1, cluster2)); |
| admin1.namespaces().setNamespaceReplicationClusters(namespace, Sets.newHashSet("r1", "r2", "r3")); |
| admin1.topics().createPartitionedTopic(partitionedTopicName, startPartitions); |
| admin1.topics().createNonPartitionedTopic(nonPartitionedTopicName); |
| |
| @Cleanup |
| PulsarClient client1 = PulsarClient.builder().serviceUrl(url1.toString()).statsInterval(0, TimeUnit.SECONDS) |
| .build(); |
| @Cleanup |
| PulsarClient client2 = PulsarClient.builder().serviceUrl(url2.toString()).statsInterval(0, TimeUnit.SECONDS) |
| .build(); |
| |
| //persistent topic |
| Producer<byte[]> persistentProducer1 = client1.newProducer().topic(partitionedTopicName).create(); |
| Producer<byte[]> persistentProducer2 = client2.newProducer().topic(partitionedTopicName).create(); |
| assertNotNull(persistentProducer1.send("test".getBytes())); |
| assertNotNull(persistentProducer2.send("test".getBytes())); |
| //non-persistent topic |
| Producer<byte[]> nonPersistentProducer1 = client1.newProducer().topic(nonPartitionedTopicName).create(); |
| Producer<byte[]> nonPersistentProducer2 = client2.newProducer().topic(nonPartitionedTopicName).create(); |
| |
| assertNotNull(nonPersistentProducer1.send("test".getBytes())); |
| assertNotNull(nonPersistentProducer2.send("test".getBytes())); |
| |
| persistentProducer1.close(); |
| persistentProducer2.close(); |
| |
| nonPersistentProducer1.close(); |
| nonPersistentProducer2.close(); |
| } |
| |
| @Test |
| public void testCleanupTopic() throws Exception { |
| |
| final String cluster1 = pulsar1.getConfig().getClusterName(); |
| final String cluster2 = pulsar2.getConfig().getClusterName(); |
| final String namespace = "pulsar/ns-" + System.nanoTime(); |
| final String topicName = "persistent://" + namespace + "/cleanTopic"; |
| final String topicMlName = namespace + "/persistent/cleanTopic"; |
| admin1.namespaces().createNamespace(namespace, Sets.newHashSet(cluster1, cluster2)); |
| |
| PulsarClient client1 = PulsarClient.builder().serviceUrl(url1.toString()).statsInterval(0, TimeUnit.SECONDS) |
| .build(); |
| |
| long topicLoadTimeoutSeconds = 3; |
| config1.setTopicLoadTimeoutSeconds(topicLoadTimeoutSeconds); |
| config2.setTopicLoadTimeoutSeconds(topicLoadTimeoutSeconds); |
| |
| ManagedLedgerFactoryImpl mlFactory = (ManagedLedgerFactoryImpl) pulsar1.getManagedLedgerClientFactory() |
| .getManagedLedgerFactory(); |
| Field ledgersField = ManagedLedgerFactoryImpl.class.getDeclaredField("ledgers"); |
| ledgersField.setAccessible(true); |
| ConcurrentHashMap<String, CompletableFuture<ManagedLedgerImpl>> ledgers = (ConcurrentHashMap<String, CompletableFuture<ManagedLedgerImpl>>) ledgersField |
| .get(mlFactory); |
| CompletableFuture<ManagedLedgerImpl> mlFuture = new CompletableFuture<>(); |
| ledgers.put(topicMlName, mlFuture); |
| |
| try { |
| Consumer<byte[]> consumer = client1.newConsumer().topic(topicName).subscriptionType(SubscriptionType.Shared) |
| .subscriptionName("my-subscriber-name").subscribeAsync().get(100, TimeUnit.MILLISECONDS); |
| fail("consumer should fail due to topic loading failure"); |
| } catch (Exception e) { |
| // Ok |
| } |
| |
| CompletableFuture<Optional<Topic>> topicFuture = null; |
| for (int i = 0; i < 5; i++) { |
| topicFuture = pulsar1.getBrokerService().getTopics().get(topicName); |
| if (topicFuture != null) { |
| break; |
| } |
| Thread.sleep(i * 1000); |
| } |
| |
| try { |
| topicFuture.get(); |
| fail("topic creation should fail"); |
| } catch (Exception e) { |
| // Ok |
| } |
| |
| final CompletableFuture<Optional<Topic>> timedOutTopicFuture = topicFuture; |
| // timeout topic future should be removed from cache |
| retryStrategically((test) -> pulsar1.getBrokerService().getTopic(topicName, false) != timedOutTopicFuture, 5, |
| 1000); |
| |
| assertNotEquals(timedOutTopicFuture, pulsar1.getBrokerService().getTopics().get(topicName)); |
| |
| try { |
| Consumer<byte[]> consumer = client1.newConsumer().topic(topicName).subscriptionType(SubscriptionType.Shared) |
| .subscriptionName("my-subscriber-name").subscribeAsync().get(100, TimeUnit.MILLISECONDS); |
| fail("consumer should fail due to topic loading failure"); |
| } catch (Exception e) { |
| // Ok |
| } |
| |
| ManagedLedgerImpl ml = (ManagedLedgerImpl) mlFactory.open(topicMlName + "-2"); |
| mlFuture.complete(ml); |
| |
| Consumer<byte[]> consumer = client1.newConsumer().topic(topicName).subscriptionName("my-subscriber-name") |
| .subscriptionType(SubscriptionType.Shared).subscribeAsync() |
| .get(2 * topicLoadTimeoutSeconds, TimeUnit.SECONDS); |
| |
| consumer.close(); |
| } |
| |
| |
| @Test |
| public void createPartitionedTopicTest() throws Exception { |
| final String cluster1 = pulsar1.getConfig().getClusterName(); |
| final String cluster2 = pulsar2.getConfig().getClusterName(); |
| final String cluster3 = pulsar3.getConfig().getClusterName(); |
| final String namespace = newUniqueName("pulsar/ns"); |
| |
| final String persistentPartitionedTopic = |
| newUniqueName("persistent://" + namespace + "/partitioned"); |
| final String persistentNonPartitionedTopic = |
| newUniqueName("persistent://" + namespace + "/non-partitioned"); |
| final String nonPersistentPartitionedTopic = |
| newUniqueName("non-persistent://" + namespace + "/partitioned"); |
| final String nonPersistentNonPartitionedTopic = |
| newUniqueName("non-persistent://" + namespace + "/non-partitioned"); |
| final int numPartitions = 3; |
| |
| admin1.namespaces().createNamespace(namespace, Sets.newHashSet(cluster1, cluster2, cluster3)); |
| admin1.namespaces().setNamespaceReplicationClusters(namespace, Sets.newHashSet("r1", "r2", "r3")); |
| |
| admin1.topics().createPartitionedTopic(persistentPartitionedTopic, numPartitions); |
| admin1.topics().createPartitionedTopic(nonPersistentPartitionedTopic, numPartitions); |
| admin1.topics().createNonPartitionedTopic(persistentNonPartitionedTopic); |
| admin1.topics().createNonPartitionedTopic(nonPersistentNonPartitionedTopic); |
| |
| List<String> partitionedTopicList = admin1.topics().getPartitionedTopicList(namespace); |
| Assert.assertTrue(partitionedTopicList.contains(persistentPartitionedTopic)); |
| Assert.assertTrue(partitionedTopicList.contains(nonPersistentPartitionedTopic)); |
| |
| // expected topic list didn't contain non-persistent-non-partitioned topic, |
| // because this model topic didn't create path in local metadata store. |
| List<String> expectedTopicList = Lists.newArrayList( |
| persistentNonPartitionedTopic, nonPersistentNonPartitionedTopic); |
| TopicName pt = TopicName.get(persistentPartitionedTopic); |
| for (int i = 0; i < numPartitions; i++) { |
| expectedTopicList.add(pt.getPartition(i).toString()); |
| } |
| |
| checkListContainExpectedTopic(admin1, namespace, expectedTopicList); |
| checkListContainExpectedTopic(admin2, namespace, expectedTopicList); |
| checkListContainExpectedTopic(admin3, namespace, expectedTopicList); |
| } |
| |
| @Test |
| public void testDoNotReplicateSystemTopic() throws Exception { |
| cleanup(); |
| config1.setTransactionCoordinatorEnabled(true); |
| config2.setTransactionCoordinatorEnabled(true); |
| config3.setTransactionCoordinatorEnabled(true); |
| setup(); |
| final String namespace = newUniqueName("pulsar/ns"); |
| admin1.namespaces().createNamespace(namespace, Sets.newHashSet("r1", "r2", "r3")); |
| String topic = TopicName.get("persistent", NamespaceName.get(namespace), |
| "testDoesNotReplicateSystemTopic").toString(); |
| String systemTopic = TopicName.get("persistent", NamespaceName.get(namespace), |
| EventsTopicNames.TRANSACTION_BUFFER_SNAPSHOT).toString(); |
| admin1.topics().createNonPartitionedTopic(topic); |
| // Replicator will not replicate System Topic other than topic policies |
| initTransaction(2, admin1, pulsar1.getBrokerServiceUrl(), pulsar1); |
| PulsarClient client = PulsarClient.builder().serviceUrl(pulsar1.getBrokerServiceUrl()) |
| .enableTransaction(true).build(); |
| TransactionImpl transaction = (TransactionImpl) client.newTransaction() |
| .withTransactionTimeout(60, TimeUnit.SECONDS).build().get(); |
| @Cleanup |
| Producer<byte[]> producer = client.newProducer().topic(topic) |
| .sendTimeout(0, TimeUnit.SECONDS) |
| .enableBatching(false).create(); |
| producer.newMessage(transaction).value("1".getBytes(StandardCharsets.UTF_8)).send(); |
| transaction.commit(); |
| |
| Awaitility.await().untilAsserted(() -> { |
| Assert.assertEquals(admin1.topics().getStats(systemTopic).getReplication().size(), 0); |
| Assert.assertEquals(admin2.topics().getStats(systemTopic).getReplication().size(), 0); |
| Assert.assertEquals(admin3.topics().getStats(systemTopic).getReplication().size(), 0); |
| }); |
| cleanup(); |
| setup(); |
| } |
| |
| private void initTransaction(int coordinatorSize, PulsarAdmin admin, String ServiceUrl, |
| PulsarService pulsarService) throws Exception { |
| admin.namespaces().createNamespace(NamespaceName.SYSTEM_NAMESPACE.toString(), coordinatorSize); |
| admin.topics().createPartitionedTopic(TopicName.TRANSACTION_COORDINATOR_ASSIGN.toString(), coordinatorSize); |
| admin.lookups().lookupTopic(TopicName.TRANSACTION_COORDINATOR_ASSIGN.toString()); |
| PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(ServiceUrl).enableTransaction(true).build(); |
| pulsarClient.close(); |
| Awaitility.await().until(() -> |
| pulsarService.getTransactionMetadataStoreService().getStores().size() == coordinatorSize); |
| } |
| |
| @Test |
| public void testLookupAnotherCluster() throws Exception { |
| log.info("--- Starting ReplicatorTest::testLookupAnotherCluster ---"); |
| |
| String namespace = "pulsar/r2/cross-cluster-ns"; |
| admin1.namespaces().createNamespace(namespace); |
| final TopicName topicName = TopicName |
| .get("persistent://" + namespace + "/topic"); |
| |
| @Cleanup |
| PulsarClient client1 = PulsarClient.builder() |
| .serviceUrl(url1.toString()).statsInterval(0, TimeUnit.SECONDS) |
| .build(); |
| Producer<byte[]> producer = client1.newProducer().topic(topicName.toString()) |
| .enableBatching(false) |
| .messageRoutingMode(MessageRoutingMode.SinglePartition) |
| .create(); |
| |
| producer.close(); |
| } |
| |
| private void checkListContainExpectedTopic(PulsarAdmin admin, String namespace, List<String> expectedTopicList) { |
| // wait non-partitioned topics replicators created finished |
| final List<String> list = new ArrayList<>(); |
| Awaitility.await().atMost(2, TimeUnit.SECONDS).until(() -> { |
| list.clear(); |
| list.addAll(admin.topics().getList(namespace).stream() |
| .filter(topic -> !topic.contains(EventsTopicNames.NAMESPACE_EVENTS_LOCAL_NAME)) |
| .collect(Collectors.toList())); |
| return list.size() == expectedTopicList.size(); |
| }); |
| for (String expectTopic : expectedTopicList) { |
| Assert.assertTrue(list.contains(expectTopic)); |
| } |
| } |
| |
| @Test |
| public void testReplicatorWithFailedAck() throws Exception { |
| |
| log.info("--- Starting ReplicatorTest::testReplication ---"); |
| |
| String namespace = "pulsar/global/ns2"; |
| admin1.namespaces().createNamespace(namespace, Sets.newHashSet("r1")); |
| final TopicName dest = TopicName |
| .get(BrokerTestUtil.newUniqueName("persistent://" + namespace + "/ackFailedTopic")); |
| |
| @Cleanup |
| MessageProducer producer1 = new MessageProducer(url1, dest); |
| |
| PersistentTopic topic = (PersistentTopic) pulsar1.getBrokerService().getTopic(dest.toString(), false) |
| .getNow(null).get(); |
| final ManagedLedgerImpl managedLedger = (ManagedLedgerImpl) topic.getManagedLedger(); |
| final ManagedCursorImpl cursor = (ManagedCursorImpl) managedLedger.openCursor("pulsar.repl.r2"); |
| final ManagedCursorImpl spyCursor = spy(cursor); |
| managedLedger.getCursors().removeCursor(cursor.getName()); |
| managedLedger.getCursors().add(spyCursor, PositionImpl.EARLIEST); |
| AtomicBoolean isMakeAckFail = new AtomicBoolean(false); |
| doAnswer(invocation -> { |
| Position pos = (Position) invocation.getArguments()[0]; |
| AsyncCallbacks.DeleteCallback cb = (AsyncCallbacks.DeleteCallback) invocation.getArguments()[1]; |
| Object ctx = invocation.getArguments()[2]; |
| if (isMakeAckFail.get()) { |
| log.info("async-delete {} will be failed", pos); |
| cb.deleteFailed(new ManagedLedgerException("mocked error"), ctx); |
| } else { |
| log.info("async-delete {} will success", pos); |
| cursor.asyncDelete(pos, cb, ctx); |
| } |
| return null; |
| }).when(spyCursor).asyncDelete(Mockito.any(Position.class), Mockito.any(AsyncCallbacks.DeleteCallback.class), |
| Mockito.any()); |
| |
| log.info("--- Starting producer --- " + url1); |
| admin1.namespaces().setNamespaceReplicationClusters(namespace, Sets.newHashSet("r1", "r2")); |
| // Produce from cluster1 and consume from the rest |
| producer1.produce(2); |
| |
| MessageIdImpl lastMessageId = (MessageIdImpl) topic.getLastMessageId().get(); |
| Position lastPosition = PositionImpl.get(lastMessageId.getLedgerId(), lastMessageId.getEntryId()); |
| ConcurrentOpenHashMap<String, Replicator> replicators = topic.getReplicators(); |
| |
| Awaitility.await().pollInterval(1, TimeUnit.SECONDS).timeout(30, TimeUnit.SECONDS) |
| .untilAsserted(() -> { |
| PersistentReplicator replicator = (PersistentReplicator) replicators.get("r2"); |
| assertNotNull(replicator); |
| assertEquals(org.apache.pulsar.broker.service.AbstractReplicator.State.Started, |
| replicator.getState()); |
| }); |
| |
| // Make sure all the data has replicated to the remote cluster before close the cursor. |
| Awaitility.await().untilAsserted(() -> assertEquals(cursor.getMarkDeletedPosition(), lastPosition)); |
| |
| isMakeAckFail.set(true); |
| |
| producer1.produce(10); |
| |
| // The cursor is closed, so the mark delete position will not move forward. |
| assertEquals(cursor.getMarkDeletedPosition(), lastPosition); |
| |
| isMakeAckFail.set(false); |
| |
| Awaitility.await().timeout(30, TimeUnit.SECONDS).until( |
| () -> { |
| log.info("++++++++++++ {}, {}", cursor.getMarkDeletedPosition(), cursor.getReadPosition()); |
| return cursor.getMarkDeletedPosition().getEntryId() == (cursor.getReadPosition().getEntryId() - 1); |
| }); |
| } |
| |
| private static final Logger log = LoggerFactory.getLogger(ReplicatorTest.class); |
| |
| @Test |
| public void testWhenUpdateReplicationCluster() throws Exception { |
| log.info("--- testWhenUpdateReplicationCluster ---"); |
| String namespace = "pulsar/ns2"; |
| admin1.namespaces().createNamespace(namespace); |
| admin1.namespaces().setNamespaceReplicationClusters(namespace, Sets.newHashSet("r1", "r2")); |
| final TopicName dest = TopicName.get( |
| BrokerTestUtil.newUniqueName("persistent://" + namespace + "/testWhenUpdateReplicationCluster")); |
| @Cleanup |
| MessageProducer producer1 = new MessageProducer(url1, dest); |
| log.info("--- Starting producer --- " + url1); |
| |
| producer1.produce(2); |
| |
| PersistentTopic topic = (PersistentTopic) pulsar1.getBrokerService().getTopic(dest.toString(), false) |
| .getNow(null).get(); |
| Awaitility.await().untilAsserted(() -> { |
| assertTrue(topic.getReplicators().containsKey("r2")); |
| }); |
| |
| admin1.topics().setReplicationClusters(dest.toString(), Lists.newArrayList("r1")); |
| |
| Awaitility.await().untilAsserted(() -> { |
| Set<String> replicationClusters = admin1.topics().getReplicationClusters(dest.toString(), false); |
| assertTrue(replicationClusters != null && replicationClusters.size() == 1); |
| assertTrue(topic.getReplicators().isEmpty()); |
| }); |
| } |
| |
| @Test |
| public void testReplicatorProducerNotExceed() throws Exception { |
| log.info("--- testReplicatorProducerNotExceed ---"); |
| String namespace1 = "pulsar/ns11"; |
| admin1.namespaces().createNamespace(namespace1); |
| admin1.namespaces().setNamespaceReplicationClusters(namespace1, Sets.newHashSet("r1", "r2")); |
| final TopicName dest1 = TopicName.get( |
| BrokerTestUtil.newUniqueName("persistent://" + namespace1 + "/testReplicatorProducerNotExceed1")); |
| String namespace2 = "pulsar/ns22"; |
| admin2.namespaces().createNamespace(namespace2); |
| admin2.namespaces().setNamespaceReplicationClusters(namespace2, Sets.newHashSet("r1", "r2")); |
| final TopicName dest2 = TopicName.get( |
| BrokerTestUtil.newUniqueName("persistent://" + namespace1 + "/testReplicatorProducerNotExceed2")); |
| admin1.topics().createPartitionedTopic(dest1.toString(), 1); |
| admin1.topicPolicies().setMaxProducers(dest1.toString(), 1); |
| admin2.topics().createPartitionedTopic(dest2.toString(), 1); |
| admin2.topicPolicies().setMaxProducers(dest2.toString(), 1); |
| @Cleanup |
| MessageProducer producer1 = new MessageProducer(url1, dest1); |
| log.info("--- Starting producer1 --- " + url1); |
| |
| producer1.produce(1); |
| |
| @Cleanup |
| MessageProducer producer2 = new MessageProducer(url2, dest2); |
| log.info("--- Starting producer2 --- " + url2); |
| |
| producer2.produce(1); |
| |
| Assert.assertThrows(PulsarClientException.ProducerBusyException.class, () -> new MessageProducer(url2, dest2)); |
| } |
| } |
