| /** |
| * 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.client.api; |
| |
| import static org.testng.Assert.assertEquals; |
| import static org.testng.Assert.assertFalse; |
| import static org.testng.Assert.assertNull; |
| import static org.testng.Assert.assertTrue; |
| import static org.testng.Assert.fail; |
| import com.google.common.collect.Lists; |
| import com.google.common.collect.Maps; |
| import com.google.common.collect.Sets; |
| import java.io.IOException; |
| import java.nio.file.Files; |
| import java.nio.file.Paths; |
| import java.util.ArrayList; |
| import java.util.Collections; |
| import java.util.HashSet; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Random; |
| import java.util.Set; |
| import java.util.concurrent.CompletableFuture; |
| import java.util.concurrent.CountDownLatch; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.atomic.AtomicReference; |
| |
| import lombok.Cleanup; |
| |
| import org.apache.bookkeeper.common.concurrent.FutureUtils; |
| import org.apache.commons.lang3.tuple.Pair; |
| import org.apache.pulsar.client.impl.BatchMessageIdImpl; |
| import org.apache.pulsar.client.impl.MessageIdImpl; |
| import org.apache.pulsar.client.impl.MessageImpl; |
| import org.apache.pulsar.client.impl.MultiTopicsConsumerImpl; |
| import org.apache.pulsar.client.impl.MultiTopicsReaderImpl; |
| import org.apache.pulsar.client.impl.ReaderImpl; |
| import org.apache.pulsar.client.impl.TopicMessageIdImpl; |
| import org.apache.pulsar.client.impl.TopicMessageImpl; |
| import org.apache.pulsar.common.policies.data.TopicStats; |
| import org.apache.pulsar.common.util.RelativeTimeUtil; |
| 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.DataProvider; |
| import org.testng.annotations.Test; |
| |
| @Test(groups = "flaky") |
| public class TopicReaderTest extends ProducerConsumerBase { |
| private static final Logger log = LoggerFactory.getLogger(TopicReaderTest.class); |
| |
| @BeforeClass |
| @Override |
| protected void setup() throws Exception { |
| super.internalSetup(); |
| super.producerBaseSetup(); |
| } |
| |
| @AfterClass(alwaysRun = true) |
| @Override |
| protected void cleanup() throws Exception { |
| super.internalCleanup(); |
| } |
| |
| @DataProvider |
| public static Object[][] variationsForExpectedPos() { |
| return new Object[][] { |
| // batching / start-inclusive / num-of-messages |
| {true, true, 10 }, |
| {true, false, 10 }, |
| {false, true, 10 }, |
| {false, false, 10 }, |
| |
| {true, true, 100 }, |
| {true, false, 100 }, |
| {false, true, 100 }, |
| {false, false, 100 }, |
| }; |
| } |
| |
| @DataProvider |
| public static Object[][] variationsForResetOnLatestMsg() { |
| return new Object[][] { |
| // start-inclusive / num-of-messages |
| {true, 20}, |
| {false, 20} |
| }; |
| } |
| |
| @DataProvider |
| public static Object[][] variationsForHasMessageAvailable() { |
| return new Object[][] { |
| // batching / start-inclusive |
| {true, true}, |
| {true, false}, |
| {false, true}, |
| {false, false}, |
| }; |
| } |
| |
| @Test |
| public void testSimpleReader() throws Exception { |
| Reader<byte[]> reader = pulsarClient.newReader().topic("persistent://my-property/my-ns/testSimpleReader") |
| .startMessageId(MessageId.earliest).create(); |
| |
| Producer<byte[]> producer = pulsarClient.newProducer().topic("persistent://my-property/my-ns/testSimpleReader") |
| .create(); |
| for (int i = 0; i < 10; i++) { |
| String message = "my-message-" + i; |
| producer.send(message.getBytes()); |
| } |
| |
| Message<byte[]> msg = null; |
| Set<String> messageSet = Sets.newHashSet(); |
| for (int i = 0; i < 10; i++) { |
| msg = reader.readNext(1, TimeUnit.SECONDS); |
| |
| String receivedMessage = new String(msg.getData()); |
| log.debug("Received message: [{}]", receivedMessage); |
| String expectedMessage = "my-message-" + i; |
| testMessageOrderAndDuplicates(messageSet, receivedMessage, expectedMessage); |
| } |
| |
| // Acknowledge the consumption of all messages at once |
| reader.close(); |
| producer.close(); |
| } |
| |
| @Test |
| public void testSimpleMultiReader() throws Exception { |
| String topic = "persistent://my-property/my-ns/testSimpleMultiReader"; |
| admin.topics().createPartitionedTopic(topic, 3); |
| |
| Reader<byte[]> reader = pulsarClient.newReader().topic(topic) |
| .startMessageId(MessageId.earliest).create(); |
| |
| Producer<byte[]> producer = pulsarClient.newProducer().topic(topic) |
| .create(); |
| for (int i = 0; i < 10; i++) { |
| String message = "my-message-" + i; |
| producer.send(message.getBytes()); |
| } |
| |
| Message<byte[]> msg = null; |
| Set<String> messageSet = Sets.newHashSet(); |
| for (int i = 0; i < 10; i++) { |
| msg = reader.readNext(1, TimeUnit.SECONDS); |
| String receivedMessage = new String(msg.getData()); |
| assertTrue(messageSet.add(receivedMessage)); |
| } |
| |
| reader.close(); |
| producer.close(); |
| } |
| |
| @Test |
| public void testReaderAfterMessagesWerePublished() throws Exception { |
| Producer<byte[]> producer = pulsarClient.newProducer().topic("persistent://my-property/my-ns/testReaderAfterMessagesWerePublished") |
| .create(); |
| for (int i = 0; i < 10; i++) { |
| String message = "my-message-" + i; |
| producer.send(message.getBytes()); |
| } |
| |
| Reader<byte[]> reader = pulsarClient.newReader() |
| .topic("persistent://my-property/my-ns/testReaderAfterMessagesWerePublished") |
| .startMessageId(MessageId.earliest).create(); |
| |
| Message<byte[]> msg = null; |
| Set<String> messageSet = Sets.newHashSet(); |
| for (int i = 0; i < 10; i++) { |
| msg = reader.readNext(1, TimeUnit.SECONDS); |
| |
| String receivedMessage = new String(msg.getData()); |
| log.debug("Received message: [{}]", receivedMessage); |
| String expectedMessage = "my-message-" + i; |
| testMessageOrderAndDuplicates(messageSet, receivedMessage, expectedMessage); |
| } |
| |
| // Acknowledge the consumption of all messages at once |
| reader.close(); |
| producer.close(); |
| } |
| |
| @Test |
| public void testMultiReaderAfterMessagesWerePublished() throws Exception { |
| String topic = "persistent://my-property/my-ns/testMultiReaderAfterMessagesWerePublished"; |
| admin.topics().createPartitionedTopic(topic, 3); |
| Producer<byte[]> producer = pulsarClient.newProducer().topic(topic) |
| .create(); |
| for (int i = 0; i < 10; i++) { |
| String message = "my-message-" + i; |
| producer.send(message.getBytes()); |
| } |
| |
| Reader<byte[]> reader = pulsarClient.newReader().topic(topic) |
| .startMessageId(MessageId.earliest).create(); |
| |
| Message<byte[]> msg = null; |
| Set<String> messageSet = Sets.newHashSet(); |
| for (int i = 0; i < 10; i++) { |
| msg = reader.readNext(1, TimeUnit.SECONDS); |
| |
| String receivedMessage = new String(msg.getData()); |
| assertTrue(messageSet.add(receivedMessage)); |
| } |
| |
| reader.close(); |
| producer.close(); |
| } |
| |
| @Test |
| public void testMultipleReaders() throws Exception { |
| Producer<byte[]> producer = pulsarClient.newProducer().topic("persistent://my-property/my-ns/testMultipleReaders") |
| .create(); |
| for (int i = 0; i < 10; i++) { |
| String message = "my-message-" + i; |
| producer.send(message.getBytes()); |
| } |
| |
| Reader<byte[]> reader1 = pulsarClient.newReader().topic("persistent://my-property/my-ns/testMultipleReaders") |
| .startMessageId(MessageId.earliest).create(); |
| |
| Reader<byte[]> reader2 = pulsarClient.newReader().topic("persistent://my-property/my-ns/testMultipleReaders") |
| .startMessageId(MessageId.earliest).create(); |
| |
| Message<byte[]> msg = null; |
| Set<String> messageSet1 = Sets.newHashSet(); |
| for (int i = 0; i < 10; i++) { |
| msg = reader1.readNext(1, TimeUnit.SECONDS); |
| |
| String receivedMessage = new String(msg.getData()); |
| log.debug("Received message: [{}]", receivedMessage); |
| String expectedMessage = "my-message-" + i; |
| testMessageOrderAndDuplicates(messageSet1, receivedMessage, expectedMessage); |
| } |
| |
| Set<String> messageSet2 = Sets.newHashSet(); |
| for (int i = 0; i < 10; i++) { |
| msg = reader2.readNext(1, TimeUnit.SECONDS); |
| |
| String receivedMessage = new String(msg.getData()); |
| log.debug("Received message: [{}]", receivedMessage); |
| String expectedMessage = "my-message-" + i; |
| testMessageOrderAndDuplicates(messageSet2, receivedMessage, expectedMessage); |
| } |
| |
| reader1.close(); |
| reader2.close(); |
| producer.close(); |
| } |
| |
| @Test |
| public void testMultiMultipleReaders() throws Exception { |
| final String topic = "persistent://my-property/my-ns/testMultiMultipleReaders"; |
| admin.topics().createPartitionedTopic(topic, 3); |
| Producer<byte[]> producer = pulsarClient.newProducer().topic(topic) |
| .create(); |
| for (int i = 0; i < 10; i++) { |
| String message = "my-message-" + i; |
| producer.send(message.getBytes()); |
| } |
| |
| Reader<byte[]> reader1 = pulsarClient.newReader().topic(topic) |
| .startMessageId(MessageId.earliest).create(); |
| |
| Reader<byte[]> reader2 = pulsarClient.newReader().topic(topic) |
| .startMessageId(MessageId.earliest).create(); |
| |
| Message<byte[]> msg = null; |
| Set<String> messageSet1 = Sets.newHashSet(); |
| for (int i = 0; i < 10; i++) { |
| msg = reader1.readNext(1, TimeUnit.SECONDS); |
| String receivedMessage = new String(msg.getData()); |
| assertTrue(messageSet1.add(receivedMessage)); |
| } |
| |
| Set<String> messageSet2 = Sets.newHashSet(); |
| for (int i = 0; i < 10; i++) { |
| msg = reader2.readNext(1, TimeUnit.SECONDS); |
| |
| String receivedMessage = new String(msg.getData()); |
| assertTrue(messageSet2.add(receivedMessage)); |
| } |
| |
| reader1.close(); |
| reader2.close(); |
| producer.close(); |
| } |
| |
| @Test |
| public void testTopicStats() throws Exception { |
| String topicName = "persistent://my-property/my-ns/testTopicStats"; |
| |
| Reader<byte[]> reader1 = pulsarClient.newReader().topic(topicName).startMessageId(MessageId.earliest).create(); |
| |
| Reader<byte[]> reader2 = pulsarClient.newReader().topic(topicName).startMessageId(MessageId.earliest).create(); |
| |
| TopicStats stats = admin.topics().getStats(topicName); |
| assertEquals(stats.getSubscriptions().size(), 2); |
| |
| reader1.close(); |
| stats = admin.topics().getStats(topicName); |
| assertEquals(stats.getSubscriptions().size(), 1); |
| |
| reader2.close(); |
| |
| stats = admin.topics().getStats(topicName); |
| assertEquals(stats.getSubscriptions().size(), 0); |
| } |
| |
| @Test |
| public void testMultiTopicStats() throws Exception { |
| String topicName = "persistent://my-property/my-ns/testMultiTopicStats"; |
| admin.topics().createPartitionedTopic(topicName, 3); |
| |
| Reader<byte[]> reader1 = pulsarClient.newReader().topic(topicName).startMessageId(MessageId.earliest).create(); |
| |
| Reader<byte[]> reader2 = pulsarClient.newReader().topic(topicName).startMessageId(MessageId.earliest).create(); |
| |
| TopicStats stats = admin.topics().getPartitionedStats(topicName,true); |
| assertEquals(stats.getSubscriptions().size(), 2); |
| |
| reader1.close(); |
| stats = admin.topics().getPartitionedStats(topicName, true); |
| assertEquals(stats.getSubscriptions().size(), 1); |
| |
| reader2.close(); |
| |
| stats = admin.topics().getPartitionedStats(topicName, true); |
| assertEquals(stats.getSubscriptions().size(), 0); |
| } |
| |
| @Test(dataProvider = "variationsForResetOnLatestMsg") |
| public void testReaderOnLatestMessage(boolean startInclusive, int numOfMessages) throws Exception { |
| final String topicName = "persistent://my-property/my-ns/ReaderOnLatestMessage"; |
| final int halfOfMsgs = numOfMessages / 2; |
| |
| Producer<byte[]> producer = pulsarClient.newProducer() |
| .topic(topicName) |
| .create(); |
| |
| for (int i = 0; i < halfOfMsgs; i++) { |
| producer.send(String.format("my-message-%d", i).getBytes()); |
| } |
| |
| ReaderBuilder<byte[]> readerBuilder = pulsarClient.newReader() |
| .topic(topicName) |
| .startMessageId(MessageId.latest); |
| |
| if (startInclusive) { |
| readerBuilder.startMessageIdInclusive(); |
| } |
| |
| Reader<byte[]> reader = readerBuilder.create(); |
| |
| for (int i = halfOfMsgs; i < numOfMessages; i++) { |
| producer.send(String.format("my-message-%d", i).getBytes()); |
| } |
| |
| // Publish more messages and verify the readers only sees new messages |
| Set<String> messageSet = Sets.newHashSet(); |
| for (int i = halfOfMsgs; i < numOfMessages; i++) { |
| Message<byte[]> message = reader.readNext(); |
| String receivedMessage = new String(message.getData()); |
| String expectedMessage = String.format("my-message-%d", i); |
| testMessageOrderAndDuplicates(messageSet, receivedMessage, expectedMessage); |
| } |
| |
| assertTrue(reader.isConnected()); |
| assertEquals(((ReaderImpl) reader).getConsumer().numMessagesInQueue(), 0); |
| assertEquals(messageSet.size(), halfOfMsgs); |
| |
| // Acknowledge the consumption of all messages at once |
| reader.close(); |
| producer.close(); |
| } |
| |
| @Test(dataProvider = "variationsForResetOnLatestMsg") |
| public void testMultiReaderOnLatestMessage(boolean startInclusive, int numOfMessages) throws Exception { |
| final String topicName = "persistent://my-property/my-ns/testMultiReaderOnLatestMessage" + System.currentTimeMillis(); |
| admin.topics().createPartitionedTopic(topicName, 3); |
| final int halfOfMsgs = numOfMessages / 2; |
| |
| Producer<byte[]> producer = pulsarClient.newProducer() |
| .topic(topicName) |
| .create(); |
| |
| Set<byte[]> oldMessage = new HashSet<>(); |
| for (int i = 0; i < halfOfMsgs; i++) { |
| byte[] message = String.format("my-message-%d", i).getBytes(); |
| producer.send(message); |
| oldMessage.add(message); |
| } |
| |
| ReaderBuilder<byte[]> readerBuilder = pulsarClient.newReader() |
| .topic(topicName) |
| .startMessageId(MessageId.latest); |
| |
| if (startInclusive) { |
| readerBuilder.startMessageIdInclusive(); |
| } |
| |
| Reader<byte[]> reader = readerBuilder.create(); |
| |
| for (int i = halfOfMsgs; i < numOfMessages; i++) { |
| producer.send(String.format("my-message-%d", i).getBytes()); |
| } |
| |
| // Publish more messages and verify the readers only sees new messages |
| Set<String> messageSet = Sets.newHashSet(); |
| for (int i = halfOfMsgs; i < numOfMessages; i++) { |
| Message<byte[]> message = reader.readNext(); |
| assertFalse(oldMessage.contains(message)); |
| String receivedMessage = new String(message.getData()); |
| assertTrue(messageSet.add(receivedMessage)); |
| } |
| |
| assertTrue(reader.isConnected()); |
| assertEquals(((MultiTopicsReaderImpl) reader).getMultiTopicsConsumer().numMessagesInQueue(), 0); |
| assertEquals(messageSet.size(), halfOfMsgs); |
| |
| producer.close(); |
| reader.close(); |
| } |
| |
| @Test |
| public void testReaderOnSpecificMessage() throws Exception { |
| Producer<byte[]> producer = pulsarClient.newProducer().topic("persistent://my-property/my-ns/testReaderOnSpecificMessage") |
| .create(); |
| List<MessageId> messageIds = new ArrayList<>(); |
| for (int i = 0; i < 10; i++) { |
| String message = "my-message-" + i; |
| messageIds.add(producer.send(message.getBytes())); |
| } |
| |
| Reader<byte[]> reader = pulsarClient.newReader().topic("persistent://my-property/my-ns/testReaderOnSpecificMessage") |
| .startMessageId(messageIds.get(4)).create(); |
| |
| // Publish more messages and verify the readers only sees messages starting from the intended message |
| Message<byte[]> msg = null; |
| Set<String> messageSet = Sets.newHashSet(); |
| for (int i = 5; i < 10; i++) { |
| msg = reader.readNext(1, TimeUnit.SECONDS); |
| |
| String receivedMessage = new String(msg.getData()); |
| log.debug("Received message: [{}]", receivedMessage); |
| String expectedMessage = "my-message-" + i; |
| testMessageOrderAndDuplicates(messageSet, receivedMessage, expectedMessage); |
| } |
| |
| // Acknowledge the consumption of all messages at once |
| reader.close(); |
| producer.close(); |
| } |
| |
| @Test |
| public void testReaderOnSpecificMessageWithBatches() throws Exception { |
| Producer<byte[]> producer = pulsarClient.newProducer() |
| .topic("persistent://my-property/my-ns/testReaderOnSpecificMessageWithBatches").enableBatching(true) |
| .batchingMaxPublishDelay(100, TimeUnit.MILLISECONDS).create(); |
| for (int i = 0; i < 10; i++) { |
| String message = "my-message-" + i; |
| producer.sendAsync(message.getBytes()); |
| } |
| |
| // Write one sync message to ensure everything before got persistend |
| producer.send("my-message-10".getBytes()); |
| Reader<byte[]> reader1 = pulsarClient.newReader() |
| .topic("persistent://my-property/my-ns/testReaderOnSpecificMessageWithBatches") |
| .startMessageId(MessageId.earliest).create(); |
| |
| MessageId lastMessageId = null; |
| for (int i = 0; i < 5; i++) { |
| Message<byte[]> msg = reader1.readNext(); |
| lastMessageId = msg.getMessageId(); |
| } |
| |
| assertEquals(lastMessageId.getClass(), BatchMessageIdImpl.class); |
| |
| System.out.println("CREATING READER ON MSG ID: " + lastMessageId); |
| |
| Reader<byte[]> reader2 = pulsarClient.newReader() |
| .topic("persistent://my-property/my-ns/testReaderOnSpecificMessageWithBatches") |
| .startMessageId(lastMessageId).create(); |
| |
| for (int i = 5; i < 11; i++) { |
| Message<byte[]> msg = reader2.readNext(1, TimeUnit.SECONDS); |
| |
| String receivedMessage = new String(msg.getData()); |
| log.debug("Received message: [{}]", receivedMessage); |
| String expectedMessage = "my-message-" + i; |
| assertEquals(receivedMessage, expectedMessage); |
| } |
| |
| producer.close(); |
| } |
| |
| @Test |
| public void testECDSAEncryption() throws Exception { |
| log.info("-- Starting {} test --", methodName); |
| |
| class EncKeyReader implements CryptoKeyReader { |
| |
| final EncryptionKeyInfo keyInfo = new EncryptionKeyInfo(); |
| |
| @Override |
| public EncryptionKeyInfo getPublicKey(String keyName, Map<String, String> keyMeta) { |
| String CERT_FILE_PATH = "./src/test/resources/certificate/public-key." + keyName; |
| if (Files.isReadable(Paths.get(CERT_FILE_PATH))) { |
| try { |
| keyInfo.setKey(Files.readAllBytes(Paths.get(CERT_FILE_PATH))); |
| return keyInfo; |
| } catch (IOException e) { |
| Assert.fail("Failed to read certificate from " + CERT_FILE_PATH); |
| } |
| } else { |
| Assert.fail("Certificate file " + CERT_FILE_PATH + " is not present or not readable."); |
| } |
| return null; |
| } |
| |
| @Override |
| public EncryptionKeyInfo getPrivateKey(String keyName, Map<String, String> keyMeta) { |
| String CERT_FILE_PATH = "./src/test/resources/certificate/private-key." + keyName; |
| if (Files.isReadable(Paths.get(CERT_FILE_PATH))) { |
| try { |
| keyInfo.setKey(Files.readAllBytes(Paths.get(CERT_FILE_PATH))); |
| return keyInfo; |
| } catch (IOException e) { |
| Assert.fail("Failed to read certificate from " + CERT_FILE_PATH); |
| } |
| } else { |
| Assert.fail("Certificate file " + CERT_FILE_PATH + " is not present or not readable."); |
| } |
| return null; |
| } |
| } |
| |
| final int totalMsg = 10; |
| |
| Set<String> messageSet = Sets.newHashSet(); |
| Reader<byte[]> reader = pulsarClient.newReader() |
| .topic("persistent://my-property/my-ns/test-reader-myecdsa-topic1").startMessageId(MessageId.latest) |
| .cryptoKeyReader(new EncKeyReader()).create(); |
| |
| Producer<byte[]> producer = pulsarClient.newProducer() |
| .topic("persistent://my-property/my-ns/test-reader-myecdsa-topic1") |
| .addEncryptionKey("client-ecdsa.pem").cryptoKeyReader(new EncKeyReader()).create(); |
| for (int i = 0; i < totalMsg; i++) { |
| String message = "my-message-" + i; |
| producer.send(message.getBytes()); |
| } |
| |
| Message<byte[]> msg = null; |
| |
| for (int i = 0; i < totalMsg; i++) { |
| msg = reader.readNext(5, TimeUnit.SECONDS); |
| String receivedMessage = new String(msg.getData()); |
| log.debug("Received message: [{}]", receivedMessage); |
| String expectedMessage = "my-message-" + i; |
| testMessageOrderAndDuplicates(messageSet, receivedMessage, expectedMessage); |
| } |
| producer.close(); |
| reader.close(); |
| log.info("-- Exiting {} test --", methodName); |
| } |
| |
| @Test |
| public void testMultiReaderECDSAEncryption() throws Exception { |
| log.info("-- Starting {} test --", methodName); |
| |
| class EncKeyReader implements CryptoKeyReader { |
| |
| final EncryptionKeyInfo keyInfo = new EncryptionKeyInfo(); |
| |
| @Override |
| public EncryptionKeyInfo getPublicKey(String keyName, Map<String, String> keyMeta) { |
| String CERT_FILE_PATH = "./src/test/resources/certificate/public-key." + keyName; |
| if (Files.isReadable(Paths.get(CERT_FILE_PATH))) { |
| try { |
| keyInfo.setKey(Files.readAllBytes(Paths.get(CERT_FILE_PATH))); |
| return keyInfo; |
| } catch (IOException e) { |
| Assert.fail("Failed to read certificate from " + CERT_FILE_PATH); |
| } |
| } else { |
| Assert.fail("Certificate file " + CERT_FILE_PATH + " is not present or not readable."); |
| } |
| return null; |
| } |
| |
| @Override |
| public EncryptionKeyInfo getPrivateKey(String keyName, Map<String, String> keyMeta) { |
| String CERT_FILE_PATH = "./src/test/resources/certificate/private-key." + keyName; |
| if (Files.isReadable(Paths.get(CERT_FILE_PATH))) { |
| try { |
| keyInfo.setKey(Files.readAllBytes(Paths.get(CERT_FILE_PATH))); |
| return keyInfo; |
| } catch (IOException e) { |
| Assert.fail("Failed to read certificate from " + CERT_FILE_PATH); |
| } |
| } else { |
| Assert.fail("Certificate file " + CERT_FILE_PATH + " is not present or not readable."); |
| } |
| return null; |
| } |
| } |
| |
| final int totalMsg = 10; |
| |
| Set<String> messageSet = Sets.newHashSet(); |
| String topic = "persistent://my-property/my-ns/test-multi-reader-myecdsa-topic1"; |
| admin.topics().createPartitionedTopic(topic, 3); |
| Reader<byte[]> reader = pulsarClient.newReader() |
| .topic(topic).startMessageId(MessageId.latest) |
| .cryptoKeyReader(new EncKeyReader()).create(); |
| |
| Producer<byte[]> producer = pulsarClient.newProducer() |
| .topic(topic) |
| .addEncryptionKey("client-ecdsa.pem").cryptoKeyReader(new EncKeyReader()).create(); |
| for (int i = 0; i < totalMsg; i++) { |
| String message = "my-message-" + i; |
| producer.send(message.getBytes()); |
| } |
| |
| Message<byte[]> msg = null; |
| |
| for (int i = 0; i < totalMsg; i++) { |
| msg = reader.readNext(5, TimeUnit.SECONDS); |
| String receivedMessage = new String(msg.getData()); |
| Assert.assertTrue(messageSet.add(receivedMessage), "Received duplicate message " + receivedMessage); |
| } |
| producer.close(); |
| reader.close(); |
| } |
| |
| @Test |
| public void testDefaultCryptoKeyReader() throws Exception { |
| final String topic = "persistent://my-property/my-ns/test-reader-default-crypto-key-reader" |
| + System.currentTimeMillis(); |
| final String ecdsaPublicKeyFile = "file:./src/test/resources/certificate/public-key.client-ecdsa.pem"; |
| final String ecdsaPrivateKeyFile = "file:./src/test/resources/certificate/private-key.client-ecdsa.pem"; |
| final String ecdsaPublicKeyData = "data:application/x-pem-file;base64,LS0tLS1CRUdJTiBQVUJMSUMgS0VZLS0tLS0KTUlIS01JR2pCZ2NxaGtqT1BRSUJNSUdYQWdFQk1Cd0dCeXFHU000OUFRRUNFUUQvLy8vOS8vLy8vLy8vLy8vLwovLy8vTURzRUVQLy8vLzMvLy8vLy8vLy8vLy8vLy93RUVPaDFlY0VRZWZROTJDU1pQQ3p1WHRNREZRQUFEZzFOCmFXNW5hSFZoVVhVTXdEcEVjOUEyZVFRaEJCWWY5MUtMaVpzdERDaGdmS1VzVzRiUFdzZzVXNi9yRThBdG9wTGQKN1hxREFoRUEvLy8vL2dBQUFBQjFvdzBia0RpaEZRSUJBUU1pQUFUcktqNlJQSEdQTktjWktJT2NjTjR0Z0VOTQpuMWR6S2pMck1aVGtKNG9BYVE9PQotLS0tLUVORCBQVUJMSUMgS0VZLS0tLS0K"; |
| final String ecdsaPrivateKeyData = "data:application/x-pem-file;base64,LS0tLS1CRUdJTiBFQyBQQVJBTUVURVJTLS0tLS0KTUlHWEFnRUJNQndHQnlxR1NNNDlBUUVDRVFELy8vLzkvLy8vLy8vLy8vLy8vLy8vTURzRUVQLy8vLzMvLy8vLwovLy8vLy8vLy8vd0VFT2gxZWNFUWVmUTkyQ1NaUEN6dVh0TURGUUFBRGcxTmFXNW5hSFZoVVhVTXdEcEVjOUEyCmVRUWhCQllmOTFLTGlac3REQ2hnZktVc1c0YlBXc2c1VzYvckU4QXRvcExkN1hxREFoRUEvLy8vL2dBQUFBQjEKb3cwYmtEaWhGUUlCQVE9PQotLS0tLUVORCBFQyBQQVJBTUVURVJTLS0tLS0KLS0tLS1CRUdJTiBFQyBQUklWQVRFIEtFWS0tLS0tCk1JSFlBZ0VCQkJEZXU5aGM4a092TDNwbCtMWVNqTHE5b0lHYU1JR1hBZ0VCTUJ3R0J5cUdTTTQ5QVFFQ0VRRC8KLy8vOS8vLy8vLy8vLy8vLy8vLy9NRHNFRVAvLy8vMy8vLy8vLy8vLy8vLy8vL3dFRU9oMWVjRVFlZlE5MkNTWgpQQ3p1WHRNREZRQUFEZzFOYVc1bmFIVmhVWFVNd0RwRWM5QTJlUVFoQkJZZjkxS0xpWnN0RENoZ2ZLVXNXNGJQCldzZzVXNi9yRThBdG9wTGQ3WHFEQWhFQS8vLy8vZ0FBQUFCMW93MGJrRGloRlFJQkFhRWtBeUlBQk9zcVBwRTgKY1k4MHB4a29nNXh3M2kyQVEweWZWM01xTXVzeGxPUW5pZ0JwCi0tLS0tRU5EIEVDIFBSSVZBVEUgS0VZLS0tLS0K"; |
| final String rsaPublicKeyFile = "file:./src/test/resources/certificate/public-key.client-rsa.pem"; |
| final String rsaPrivateKeyFile = "file:./src/test/resources/certificate/private-key.client-rsa.pem"; |
| final String rsaPublicKeyData = "data:application/x-pem-file;base64,LS0tLS1CRUdJTiBQVUJMSUMgS0VZLS0tLS0KTUlJQklqQU5CZ2txaGtpRzl3MEJBUUVGQUFPQ0FROEFNSUlCQ2dLQ0FRRUF0S1d3Z3FkblRZck9DditqMU1rVApXZlNIMHdDc0haWmNhOXdBVzNxUDR1dWhsQnZuYjEwSmNGZjVaanpQOUJTWEsrdEhtSTh1b04zNjh2RXY2eWhVClJITTR5dVhxekN4enVBd2tRU28zOXJ6WDhQR0M3cWRqQ043TERKM01ucWlCSXJVc1NhRVAxd3JOc0Ixa0krbzkKRVIxZTVPL3VFUEFvdFA5MzNoSFEwSjJoTUVla0hxTDdzQmxKOThoNk5tc2ljRWFVa2FyZGswVE9YcmxrakMrYwpNZDhaYkdTY1BxSTlNMzhibW4zT0x4RlRuMXZ0aHB2blhMdkNtRzRNKzZ4dFl0RCtucGNWUFp3MWkxUjkwZk1zCjdwcFpuUmJ2OEhjL0RGZE9LVlFJZ2FtNkNEZG5OS2dXN2M3SUJNclAwQUVtMzdIVHUwTFNPalAyT0hYbHZ2bFEKR1FJREFRQUIKLS0tLS1FTkQgUFVCTElDIEtFWS0tLS0tCg=="; |
| final String rsaPrivateKeyData = "data:application/x-pem-file;base64,LS0tLS1CRUdJTiBSU0EgUFJJVkFURSBLRVktLS0tLQpNSUlFb3dJQkFBS0NBUUVBdEtXd2dxZG5UWXJPQ3YrajFNa1RXZlNIMHdDc0haWmNhOXdBVzNxUDR1dWhsQnZuCmIxMEpjRmY1Wmp6UDlCU1hLK3RIbUk4dW9OMzY4dkV2NnloVVJITTR5dVhxekN4enVBd2tRU28zOXJ6WDhQR0MKN3FkakNON0xESjNNbnFpQklyVXNTYUVQMXdyTnNCMWtJK285RVIxZTVPL3VFUEFvdFA5MzNoSFEwSjJoTUVlawpIcUw3c0JsSjk4aDZObXNpY0VhVWthcmRrMFRPWHJsa2pDK2NNZDhaYkdTY1BxSTlNMzhibW4zT0x4RlRuMXZ0Cmhwdm5YTHZDbUc0TSs2eHRZdEQrbnBjVlBadzFpMVI5MGZNczdwcFpuUmJ2OEhjL0RGZE9LVlFJZ2FtNkNEZG4KTktnVzdjN0lCTXJQMEFFbTM3SFR1MExTT2pQMk9IWGx2dmxRR1FJREFRQUJBb0lCQUFhSkZBaTJDN3UzY05yZgpBc3RZOXZWRExvTEl2SEZabGtCa3RqS1pEWW1WSXNSYitoU0NWaXdWVXJXTEw2N1I2K0l2NGVnNERlVE9BeDAwCjhwbmNYS2daVHcyd0liMS9RalIvWS9SamxhQzhsa2RtUldsaTd1ZE1RQ1pWc3lodVNqVzZQajd2cjhZRTR3b2oKRmhOaWp4RUdjZjl3V3JtTUpyemRuVFdRaVhCeW8rZVR2VVE5QlBnUEdyUmpzTVptVGtMeUFWSmZmMkRmeE81YgpJV0ZEWURKY3lZQU1DSU1RdTd2eXMvSTUwb3U2aWxiMUNPNlFNNlo3S3BQZU9vVkZQd3R6Ymg4Y2Y5eE04VU5TCmo2Si9KbWRXaGdJMzRHUzNOQTY4eFRRNlBWN3pqbmhDYytpY2NtM0pLeXpHWHdhQXBBWitFb2NlLzlqNFdLbXUKNUI0emlSMENnWUVBM2wvOU9IYmwxem15VityUnhXT0lqL2kyclR2SHp3Qm5iblBKeXVlbUw1Vk1GZHBHb2RRMwp2d0h2eVFtY0VDUlZSeG1Yb2pRNFF1UFBIczNxcDZ3RUVGUENXeENoTFNUeGxVYzg1U09GSFdVMk85OWpWN3pJCjcrSk9wREsvTXN0c3g5bkhnWGR1SkYrZ2xURnRBM0xIOE9xeWx6dTJhRlBzcHJ3S3VaZjk0UThDZ1lFQXovWngKYWtFRytQRU10UDVZUzI4Y1g1WGZqc0lYL1YyNkZzNi9zSDE2UWpVSUVkZEU1VDRmQ3Vva3hDalNpd1VjV2htbApwSEVKNVM1eHAzVllSZklTVzNqUlczcXN0SUgxdHBaaXBCNitTMHpUdUptTEpiQTNJaVdFZzJydE10N1gxdUp2CkEvYllPcWUwaE9QVHVYdVpkdFZaMG5NVEtrN0dHOE82VmtCSTdGY0NnWUVBa0RmQ21zY0pnczdKYWhsQldIbVgKekg5cHdlbStTUEtqSWMvNE5CNk4rZGdpa3gyUHAwNWhwUC9WaWhVd1lJdWZ2cy9MTm9nVllOUXJ0SGVwVW5yTgoyK1RtYkhiWmdOU3YxTGR4dDgyVWZCN3kwRnV0S3U2bGhtWEh5TmVjaG8zRmk4c2loMFYwYWlTV21ZdUhmckFICkdhaXNrRVpLbzFpaVp2UVhKSXg5TzJNQ2dZQVRCZjByOWhUWU10eXh0YzZIMy9zZGQwMUM5dGhROGdEeTB5alAKMFRxYzBkTVNKcm9EcW1JV2tvS1lldzkvYmhGQTRMVzVUQ25Xa0NBUGJIbU50RzRmZGZiWXdta0gvaGRuQTJ5MApqS2RscGZwOEdYZVVGQUdIR3gxN0ZBM3NxRnZnS1VoMGVXRWdSSFVMN3ZkUU1WRkJnSlM5M283elFNOTRmTGdQCjZjT0I4d0tCZ0ZjR1Y0R2pJMld3OWNpbGxhQzU1NE12b1NqZjhCLyswNGtYekRPaDhpWUlJek85RVVpbDFqaksKSnZ4cDRobkx6VEtXYnV4M01FV3F1ckxrWWFzNkdwS0JqdytpTk9DYXI2WWRxV0dWcU0zUlV4N1BUVWFad2tLeApVZFA2M0lmWTdpWkNJVC9RYnlIUXZJVWUyTWFpVm5IK3VseGRrSzZZNWU3Z3hjYmNrSUg0Ci0tLS0tRU5EIFJTQSBQUklWQVRFIEtFWS0tLS0tCg=="; |
| final int numMsg = 10; |
| |
| Map<String, String> privateKeyFileMap = Maps.newHashMap(); |
| privateKeyFileMap.put("client-ecdsa.pem", ecdsaPrivateKeyFile); |
| privateKeyFileMap.put("client-rsa.pem", rsaPrivateKeyFile); |
| Map<String, String> privateKeyDataMap = Maps.newHashMap(); |
| privateKeyDataMap.put("client-ecdsa.pem", ecdsaPrivateKeyData); |
| privateKeyDataMap.put("client-rsa.pem", rsaPrivateKeyData); |
| |
| Reader<byte[]> reader1 = pulsarClient.newReader().topic(topic).startMessageId(MessageId.latest) |
| .defaultCryptoKeyReader(ecdsaPrivateKeyFile).create(); |
| Reader<byte[]> reader2 = pulsarClient.newReader().topic(topic).startMessageId(MessageId.latest) |
| .defaultCryptoKeyReader(ecdsaPrivateKeyData).create(); |
| Reader<byte[]> reader3 = pulsarClient.newReader().topic(topic).startMessageId(MessageId.latest) |
| .defaultCryptoKeyReader(privateKeyFileMap).create(); |
| Reader<byte[]> reader4 = pulsarClient.newReader().topic(topic).startMessageId(MessageId.latest) |
| .defaultCryptoKeyReader(privateKeyDataMap).create(); |
| |
| Producer<byte[]> producer1 = pulsarClient.newProducer().topic(topic).addEncryptionKey("client-ecdsa.pem") |
| .defaultCryptoKeyReader(ecdsaPublicKeyFile).create(); |
| Producer<byte[]> producer2 = pulsarClient.newProducer().topic(topic).addEncryptionKey("client-ecdsa.pem") |
| .defaultCryptoKeyReader(ecdsaPublicKeyData).create(); |
| |
| for (int i = 0; i < numMsg; i++) { |
| producer1.send(("my-message-" + i).getBytes()); |
| } |
| for (int i = numMsg; i < numMsg * 2; i++) { |
| producer2.send(("my-message-" + i).getBytes()); |
| } |
| |
| producer1.close(); |
| producer2.close(); |
| |
| for (Reader<byte[]> reader : (List<Reader<byte[]>>) Lists.newArrayList(reader1, reader2)) { |
| for (int i = 0; i < numMsg * 2; i++) { |
| MessageImpl<byte[]> msg = (MessageImpl<byte[]>) reader.readNext(5, TimeUnit.SECONDS); |
| // verify that encrypted message contains encryption-context |
| msg.getEncryptionCtx().orElseThrow( |
| () -> new IllegalStateException("encryption-ctx not present for encrypted message")); |
| assertEquals(new String(msg.getData()), "my-message-" + i); |
| } |
| } |
| |
| reader1.close(); |
| reader2.close(); |
| |
| Producer<byte[]> producer3 = pulsarClient.newProducer().topic(topic).addEncryptionKey("client-rsa.pem") |
| .defaultCryptoKeyReader(rsaPublicKeyFile).create(); |
| Producer<byte[]> producer4 = pulsarClient.newProducer().topic(topic).addEncryptionKey("client-rsa.pem") |
| .defaultCryptoKeyReader(rsaPublicKeyData).create(); |
| |
| for (int i = numMsg * 2; i < numMsg * 3; i++) { |
| producer3.send(("my-message-" + i).getBytes()); |
| } |
| for (int i = numMsg * 3; i < numMsg * 4; i++) { |
| producer4.send(("my-message-" + i).getBytes()); |
| } |
| |
| producer3.close(); |
| producer4.close(); |
| |
| for (Reader<byte[]> reader : (List<Reader<byte[]>>) Lists.newArrayList(reader3, reader4)) { |
| for (int i = 0; i < numMsg * 4; i++) { |
| MessageImpl<byte[]> msg = (MessageImpl<byte[]>) reader.readNext(5, TimeUnit.SECONDS); |
| // verify that encrypted message contains encryption-context |
| msg.getEncryptionCtx().orElseThrow( |
| () -> new IllegalStateException("encryption-ctx not present for encrypted message")); |
| assertEquals(new String(msg.getData()), "my-message-" + i); |
| } |
| } |
| |
| reader3.close(); |
| reader4.close(); |
| } |
| |
| @Test |
| public void testSimpleReaderReachEndOfTopic() throws Exception { |
| Reader<byte[]> reader = pulsarClient.newReader() |
| .topic("persistent://my-property/my-ns/testSimpleReaderReachEndOfTopic") |
| .startMessageId(MessageId.earliest).create(); |
| Producer<byte[]> producer = pulsarClient.newProducer() |
| .topic("persistent://my-property/my-ns/testSimpleReaderReachEndOfTopic") |
| .create(); |
| |
| // no data write, should return false |
| assertFalse(reader.hasMessageAvailable()); |
| |
| // produce message 0 -- 99 |
| for (int i = 0; i < 100; i++) { |
| String message = "my-message-" + i; |
| producer.send(message.getBytes()); |
| } |
| |
| MessageImpl<byte[]> msg = null; |
| Set<String> messageSet = Sets.newHashSet(); |
| int index = 0; |
| |
| // read message till end. |
| while (reader.hasMessageAvailable()) { |
| msg = (MessageImpl<byte[]>) reader.readNext(1, TimeUnit.SECONDS); |
| String receivedMessage = new String(msg.getData()); |
| log.debug("Received message: [{}]", receivedMessage); |
| String expectedMessage = "my-message-" + (index++); |
| testMessageOrderAndDuplicates(messageSet, receivedMessage, expectedMessage); |
| } |
| |
| assertEquals(index, 100); |
| // readNext should return null, after reach the end of topic. |
| assertNull(reader.readNext(1, TimeUnit.SECONDS)); |
| |
| // produce message again. |
| for (int i = 100; i < 200; i++) { |
| String message = "my-message-" + i; |
| producer.send(message.getBytes()); |
| } |
| |
| // read message till end again. |
| while (reader.hasMessageAvailable()) { |
| msg = (MessageImpl<byte[]>) reader.readNext(1, TimeUnit.SECONDS); |
| String receivedMessage = new String(msg.getData()); |
| log.debug("Received message: [{}]", receivedMessage); |
| String expectedMessage = "my-message-" + (index++); |
| testMessageOrderAndDuplicates(messageSet, receivedMessage, expectedMessage); |
| } |
| |
| assertEquals(index, 200); |
| // readNext should return null, after reach the end of topic. |
| assertNull(reader.readNext(1, TimeUnit.SECONDS)); |
| |
| reader.close(); |
| producer.close(); |
| } |
| |
| @Test |
| public void testSimpleMultiReaderReachEndOfTopic() throws Exception { |
| String topic = "persistent://my-property/my-ns/testSimpleMultiReaderReachEndOfTopic"; |
| admin.topics().createPartitionedTopic(topic,3); |
| Reader<byte[]> reader = pulsarClient.newReader().topic(topic).startMessageId(MessageId.earliest).create(); |
| Producer<byte[]> producer = pulsarClient.newProducer().topic(topic).create(); |
| |
| // no data write, should return false |
| assertFalse(reader.hasMessageAvailable()); |
| |
| // produce message 0 -- 99 |
| for (int i = 0; i < 100; i++) { |
| String message = "my-message-" + i; |
| producer.send(message.getBytes()); |
| } |
| |
| TopicMessageImpl<byte[]> msg = null; |
| Set<String> messageSet = Sets.newHashSet(); |
| int index = 0; |
| |
| // read message till end. |
| while (reader.hasMessageAvailable()) { |
| msg = (TopicMessageImpl<byte[]>) reader.readNext(1, TimeUnit.SECONDS); |
| String receivedMessage = new String(msg.getData()); |
| index++; |
| Assert.assertTrue(messageSet.add(receivedMessage), "Received duplicate message " + receivedMessage); |
| } |
| |
| assertEquals(index, 100); |
| // readNext should return null, after reach the end of topic. |
| assertNull(reader.readNext(1, TimeUnit.SECONDS)); |
| |
| // produce message again. |
| for (int i = 100; i < 200; i++) { |
| String message = "my-message-" + i; |
| producer.send(message.getBytes()); |
| } |
| |
| // read message till end again. |
| while (reader.hasMessageAvailable()) { |
| msg = (TopicMessageImpl<byte[]>) reader.readNext(1, TimeUnit.SECONDS); |
| String receivedMessage = new String(msg.getData()); |
| index++; |
| Assert.assertTrue(messageSet.add(receivedMessage), "Received duplicate message " + receivedMessage); |
| } |
| |
| assertEquals(index, 200); |
| // readNext should return null, after reach the end of topic. |
| assertNull(reader.readNext(1, TimeUnit.SECONDS)); |
| |
| reader.close(); |
| producer.close(); |
| } |
| |
| @Test |
| public void testReaderReachEndOfTopicOnMessageWithBatches() throws Exception { |
| Reader<byte[]> reader = pulsarClient.newReader() |
| .topic("persistent://my-property/my-ns/testReaderReachEndOfTopicOnMessageWithBatches") |
| .startMessageId(MessageId.earliest).create(); |
| |
| Producer<byte[]> producer = pulsarClient.newProducer() |
| .topic("persistent://my-property/my-ns/testReaderReachEndOfTopicOnMessageWithBatches") |
| .enableBatching(true).batchingMaxPublishDelay(100, TimeUnit.MILLISECONDS).create(); |
| |
| // no data write, should return false |
| assertFalse(reader.hasMessageAvailable()); |
| |
| for (int i = 0; i < 100; i++) { |
| String message = "my-message-" + i; |
| producer.sendAsync(message.getBytes()); |
| } |
| |
| // Write one sync message to ensure everything before got persistend |
| producer.send("my-message-10".getBytes()); |
| |
| MessageId lastMessageId = null; |
| int index = 0; |
| assertTrue(reader.hasMessageAvailable()); |
| |
| if (reader.hasMessageAvailable()) { |
| Message<byte[]> msg = reader.readNext(); |
| lastMessageId = msg.getMessageId(); |
| assertEquals(lastMessageId.getClass(), BatchMessageIdImpl.class); |
| |
| while (msg != null) { |
| index++; |
| msg = reader.readNext(100, TimeUnit.MILLISECONDS); |
| } |
| assertEquals(index, 101); |
| } |
| |
| assertFalse(reader.hasMessageAvailable()); |
| |
| reader.close(); |
| producer.close(); |
| } |
| |
| @Test |
| public void testMultiReaderReachEndOfTopicOnMessageWithBatches() throws Exception { |
| String topic = "persistent://my-property/my-ns/testMultiReaderReachEndOfTopicOnMessageWithBatches"; |
| admin.topics().createPartitionedTopic(topic, 3); |
| Reader<byte[]> reader = pulsarClient.newReader() |
| .topic(topic) |
| .startMessageId(MessageId.earliest).create(); |
| |
| Producer<byte[]> producer = pulsarClient.newProducer() |
| .topic(topic) |
| .enableBatching(true).batchingMaxPublishDelay(100, TimeUnit.MILLISECONDS).create(); |
| |
| // no data write, should return false |
| assertFalse(reader.hasMessageAvailable()); |
| |
| for (int i = 0; i < 100; i++) { |
| String message = "my-message-" + i; |
| producer.sendAsync(message.getBytes()); |
| } |
| |
| // Write one sync message to ensure everything before got persistend |
| producer.send("my-message-10".getBytes()); |
| |
| MessageId lastMessageId = null; |
| int index = 0; |
| assertTrue(reader.hasMessageAvailable()); |
| |
| if (reader.hasMessageAvailable()) { |
| Message<byte[]> msg = reader.readNext(); |
| lastMessageId = msg.getMessageId(); |
| assertEquals(lastMessageId.getClass(), TopicMessageIdImpl.class); |
| |
| while (msg != null) { |
| index++; |
| msg = reader.readNext(100, TimeUnit.MILLISECONDS); |
| } |
| assertEquals(index, 101); |
| } |
| |
| assertFalse(reader.hasMessageAvailable()); |
| |
| reader.close(); |
| producer.close(); |
| } |
| |
| @Test |
| public void testMessageAvailableAfterRestart() throws Exception { |
| String topic = "persistent://my-property/use/my-ns/testMessageAvailableAfterRestart"; |
| String content = "my-message-1"; |
| |
| // stop retention from cleaning up |
| pulsarClient.newConsumer().topic(topic).subscriptionName("sub1").subscribe().close(); |
| |
| try (Reader<byte[]> reader = pulsarClient.newReader().topic(topic) |
| .startMessageId(MessageId.earliest).create()) { |
| assertFalse(reader.hasMessageAvailable()); |
| } |
| |
| try (Producer<byte[]> producer = pulsarClient.newProducer().topic(topic).create()) { |
| producer.send(content.getBytes()); |
| } |
| |
| try (Reader<byte[]> reader = pulsarClient.newReader().topic(topic) |
| .startMessageId(MessageId.earliest).create()) { |
| assertTrue(reader.hasMessageAvailable()); |
| } |
| |
| // cause broker to drop topic. Will be loaded next time we access it |
| pulsar.getBrokerService().getTopicReference(topic).get().close(false).get(); |
| |
| try (Reader<byte[]> reader = pulsarClient.newReader().topic(topic) |
| .startMessageId(MessageId.earliest).create()) { |
| assertTrue(reader.hasMessageAvailable()); |
| |
| String readOut = new String(reader.readNext().getData()); |
| assertEquals(content, readOut); |
| assertFalse(reader.hasMessageAvailable()); |
| } |
| |
| } |
| |
| @Test |
| public void testMultiReaderMessageAvailableAfterRestart() throws Exception { |
| String topic = "persistent://my-property/use/my-ns/testMessageAvailableAfterRestart2"; |
| String content = "my-message-1"; |
| admin.topics().createPartitionedTopic(topic, 3); |
| // stop retention from cleaning up |
| pulsarClient.newConsumer().topic(topic).subscriptionName("sub2").subscribe().close(); |
| |
| try (Reader<byte[]> reader = pulsarClient.newReader().topic(topic) |
| .startMessageId(MessageId.earliest).create()) { |
| assertFalse(reader.hasMessageAvailable()); |
| } |
| |
| try (Producer<byte[]> producer = pulsarClient.newProducer().topic(topic).create()) { |
| producer.send(content.getBytes()); |
| } |
| |
| try (Reader<byte[]> reader = pulsarClient.newReader().topic(topic) |
| .startMessageId(MessageId.earliest).create()) { |
| assertTrue(reader.hasMessageAvailable()); |
| } |
| |
| // cause broker to drop topic. Will be loaded next time we access it |
| pulsar.getBrokerService().getTopics().keys().forEach(topicName -> { |
| try { |
| pulsar.getBrokerService().getTopicReference(topicName).get().close(false).get(); |
| } catch (Exception e) { |
| fail(); |
| } |
| }); |
| |
| try (Reader<byte[]> reader = pulsarClient.newReader().topic(topic) |
| .startMessageId(MessageId.earliest).create()) { |
| assertTrue(reader.hasMessageAvailable()); |
| |
| String readOut = new String(reader.readNext().getData()); |
| assertEquals(content, readOut); |
| assertFalse(reader.hasMessageAvailable()); |
| } |
| |
| } |
| |
| @Test(dataProvider = "variationsForHasMessageAvailable") |
| public void testHasMessageAvailable(boolean enableBatch, boolean startInclusive) throws Exception { |
| final String topicName = "persistent://my-property/my-ns/HasMessageAvailable"; |
| final int numOfMessage = 100; |
| |
| ProducerBuilder<byte[]> producerBuilder = pulsarClient.newProducer() |
| .topic(topicName); |
| |
| if (enableBatch) { |
| producerBuilder |
| .enableBatching(true) |
| .batchingMaxMessages(10); |
| } else { |
| producerBuilder |
| .enableBatching(false); |
| } |
| |
| Producer<byte[]> producer = producerBuilder.create(); |
| |
| CountDownLatch latch = new CountDownLatch(numOfMessage); |
| |
| List<MessageId> allIds = Collections.synchronizedList(new ArrayList<>()); |
| |
| for (int i = 0; i < numOfMessage; i++) { |
| producer.sendAsync(String.format("msg num %d", i).getBytes()).whenComplete((mid, e) -> { |
| if (e != null) { |
| Assert.fail(); |
| } else { |
| allIds.add(mid); |
| } |
| latch.countDown(); |
| }); |
| } |
| |
| latch.await(); |
| |
| allIds.sort(null); // make sure the largest mid appears at last. |
| |
| for (MessageId id : allIds) { |
| Reader<byte[]> reader; |
| |
| if (startInclusive) { |
| reader = pulsarClient.newReader().topic(topicName) |
| .startMessageId(id).startMessageIdInclusive().create(); |
| } else { |
| reader = pulsarClient.newReader().topic(topicName) |
| .startMessageId(id).create(); |
| } |
| if (startInclusive) { |
| assertTrue(reader.hasMessageAvailable()); |
| } else if (id != allIds.get(allIds.size() - 1)) { |
| assertTrue(reader.hasMessageAvailable()); |
| } else { |
| assertFalse(reader.hasMessageAvailable()); |
| } |
| reader.close(); |
| } |
| |
| producer.close(); |
| } |
| |
| @Test(timeOut = 20000) |
| public void testHasMessageAvailableWithBatch() throws Exception { |
| final String topicName = "persistent://my-property/my-ns/testHasMessageAvailableWithBatch"; |
| final int numOfMessage = 10; |
| |
| Producer<byte[]> producer = pulsarClient.newProducer() |
| .enableBatching(true) |
| .batchingMaxMessages(10) |
| .batchingMaxPublishDelay(2,TimeUnit.SECONDS) |
| .topic(topicName).create(); |
| |
| //For batch-messages with single message, the type of client messageId should be the same as that of broker |
| MessageIdImpl messageId = (MessageIdImpl) producer.send("msg".getBytes()); |
| assertTrue(messageId instanceof MessageIdImpl); |
| ReaderImpl<byte[]> reader = (ReaderImpl<byte[]>)pulsarClient.newReader().topic(topicName) |
| .startMessageId(messageId).startMessageIdInclusive().create(); |
| MessageIdImpl lastMsgId = (MessageIdImpl) reader.getConsumer().getLastMessageId(); |
| assertTrue(messageId instanceof BatchMessageIdImpl); |
| assertEquals(lastMsgId.getLedgerId(), messageId.getLedgerId()); |
| assertEquals(lastMsgId.getEntryId(), messageId.getEntryId()); |
| reader.close(); |
| |
| CountDownLatch latch = new CountDownLatch(numOfMessage); |
| List<MessageId> allIds = Collections.synchronizedList(new ArrayList<>()); |
| for (int i = 0; i < numOfMessage; i++) { |
| producer.sendAsync(String.format("msg num %d", i).getBytes()).whenComplete((mid, e) -> { |
| if (e != null) { |
| Assert.fail(); |
| } else { |
| allIds.add(mid); |
| } |
| latch.countDown(); |
| }); |
| } |
| producer.flush(); |
| latch.await(); |
| producer.close(); |
| |
| //For batch-message with multi messages, the type of client messageId should be the same as that of broker |
| for (MessageId id : allIds) { |
| reader = (ReaderImpl<byte[]>) pulsarClient.newReader().topic(topicName) |
| .startMessageId(id).startMessageIdInclusive().create(); |
| if (id instanceof BatchMessageIdImpl) { |
| MessageId lastMessageId = reader.getConsumer().getLastMessageId(); |
| assertTrue(lastMessageId instanceof BatchMessageIdImpl); |
| log.info("id {} instance of BatchMessageIdImpl",id); |
| } else { |
| assertTrue(id instanceof MessageIdImpl); |
| MessageId lastMessageId = reader.getConsumer().getLastMessageId(); |
| assertTrue(lastMessageId instanceof MessageIdImpl); |
| log.info("id {} instance of MessageIdImpl",id); |
| } |
| reader.close(); |
| } |
| //For non-batch message, the type of client messageId should be the same as that of broker |
| producer = pulsarClient.newProducer() |
| .enableBatching(false).topic(topicName).create(); |
| messageId = (MessageIdImpl) producer.send("non-batch".getBytes()); |
| assertFalse(messageId instanceof BatchMessageIdImpl); |
| assertTrue(messageId instanceof MessageIdImpl); |
| reader = (ReaderImpl<byte[]>) pulsarClient.newReader().topic(topicName) |
| .startMessageId(messageId).create(); |
| MessageId lastMessageId = reader.getConsumer().getLastMessageId(); |
| assertFalse(lastMessageId instanceof BatchMessageIdImpl); |
| assertTrue(lastMessageId instanceof MessageIdImpl); |
| assertEquals(lastMessageId, messageId); |
| producer.close(); |
| reader.close(); |
| } |
| |
| @Test |
| public void testReaderNonDurableIsAbleToSeekRelativeTime() throws Exception { |
| final int numOfMessage = 10; |
| final String topicName = "persistent://my-property/my-ns/ReaderNonDurableIsAbleToSeekRelativeTime"; |
| |
| Producer<byte[]> producer = pulsarClient.newProducer() |
| .topic(topicName).create(); |
| |
| for (int i = 0; i < numOfMessage; i++) { |
| producer.send(String.format("msg num %d", i).getBytes()); |
| } |
| |
| Reader<byte[]> reader = pulsarClient.newReader().topic(topicName) |
| .startMessageId(MessageId.earliest).create(); |
| assertTrue(reader.hasMessageAvailable()); |
| |
| reader.seek(RelativeTimeUtil.parseRelativeTimeInSeconds("-1m")); |
| |
| assertTrue(reader.hasMessageAvailable()); |
| |
| reader.close(); |
| producer.close(); |
| } |
| |
| @Test |
| public void testMultiReaderNonDurableIsAbleToSeekRelativeTime() throws Exception { |
| final int numOfMessage = 10; |
| final String topicName = "persistent://my-property/my-ns/ReaderNonDurableIsAbleToSeekRelativeTime"; |
| admin.topics().createPartitionedTopic(topicName, 3); |
| |
| Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName).create(); |
| |
| for (int i = 0; i < numOfMessage; i++) { |
| producer.send(String.format("msg num %d", i).getBytes()); |
| } |
| |
| Reader<byte[]> reader = pulsarClient.newReader().topic(topicName).startMessageId(MessageId.earliest).create(); |
| assertTrue(reader.hasMessageAvailable()); |
| |
| reader.seek(RelativeTimeUtil.parseRelativeTimeInSeconds("-1m")); |
| |
| assertTrue(reader.hasMessageAvailable()); |
| |
| reader.close(); |
| producer.close(); |
| } |
| |
| @Test |
| public void testReaderIsAbleToSeekWithTimeOnBeginningOfTopic() throws Exception { |
| final String topicName = "persistent://my-property/my-ns/ReaderSeekWithTimeOnBeginningOfTopic"; |
| final int numOfMessage = 10; |
| |
| Producer<byte[]> producer = pulsarClient.newProducer() |
| .topic(topicName).create(); |
| |
| for (int i = 0; i < numOfMessage; i++) { |
| producer.send(String.format("msg num %d", i).getBytes()); |
| } |
| |
| Reader<byte[]> reader = pulsarClient.newReader().topic(topicName) |
| .startMessageId(MessageId.earliest).create(); |
| |
| assertTrue(reader.hasMessageAvailable()); |
| |
| // Read all messages the first time |
| Set<String> messageSetA = Sets.newHashSet(); |
| for (int i = 0; i < numOfMessage; i++) { |
| Message<byte[]> message = reader.readNext(); |
| String receivedMessage = new String(message.getData()); |
| String expectedMessage = String.format("msg num %d", i); |
| testMessageOrderAndDuplicates(messageSetA, receivedMessage, expectedMessage); |
| } |
| |
| assertFalse(reader.hasMessageAvailable()); |
| |
| // Perform cursor reset by time |
| reader.seek(RelativeTimeUtil.parseRelativeTimeInSeconds("-1m")); |
| |
| // Read all messages a second time after seek() |
| Set<String> messageSetB = Sets.newHashSet(); |
| for (int i = 0; i < numOfMessage; i++) { |
| Message<byte[]> message = reader.readNext(); |
| String receivedMessage = new String(message.getData()); |
| String expectedMessage = String.format("msg num %d", i); |
| testMessageOrderAndDuplicates(messageSetB, receivedMessage, expectedMessage); |
| } |
| |
| // Reader should be finished |
| assertTrue(reader.isConnected()); |
| assertFalse(reader.hasMessageAvailable()); |
| assertEquals(((ReaderImpl) reader).getConsumer().numMessagesInQueue(), 0); |
| |
| reader.close(); |
| producer.close(); |
| } |
| |
| @Test |
| public void testMultiReaderIsAbleToSeekWithTimeOnBeginningOfTopic() throws Exception { |
| final String topicName = "persistent://my-property/my-ns/MultiReaderSeekWithTimeOnBeginningOfTopic"; |
| final int numOfMessage = 10; |
| admin.topics().createPartitionedTopic(topicName, 3); |
| |
| Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName).create(); |
| |
| for (int i = 0; i < numOfMessage; i++) { |
| producer.send(String.format("msg num %d", i).getBytes()); |
| } |
| |
| Reader<byte[]> reader = pulsarClient.newReader().topic(topicName).startMessageId(MessageId.earliest).create(); |
| |
| assertTrue(reader.hasMessageAvailable()); |
| |
| // Read all messages the first time |
| Set<String> messageSetA = Sets.newHashSet(); |
| for (int i = 0; i < numOfMessage; i++) { |
| Message<byte[]> message = reader.readNext(); |
| String receivedMessage = new String(message.getData()); |
| Assert.assertTrue(messageSetA.add(receivedMessage), "Received duplicate message " + receivedMessage); |
| } |
| |
| assertFalse(reader.hasMessageAvailable()); |
| |
| // Perform cursor reset by time |
| reader.seek(RelativeTimeUtil.parseRelativeTimeInSeconds("-1m")); |
| |
| // Read all messages a second time after seek() |
| Set<String> messageSetB = Sets.newHashSet(); |
| for (int i = 0; i < numOfMessage; i++) { |
| Message<byte[]> message = reader.readNext(); |
| String receivedMessage = new String(message.getData()); |
| Assert.assertTrue(messageSetB.add(receivedMessage), "Received duplicate message " + receivedMessage); |
| } |
| |
| // Reader should be finished |
| assertTrue(reader.isConnected()); |
| assertFalse(reader.hasMessageAvailable()); |
| assertEquals(((MultiTopicsReaderImpl) reader).getMultiTopicsConsumer().numMessagesInQueue(), 0); |
| |
| reader.close(); |
| producer.close(); |
| } |
| |
| @Test |
| public void testReaderIsAbleToSeekWithMessageIdOnMiddleOfTopic() throws Exception { |
| final String topicName = "persistent://my-property/my-ns/ReaderSeekWithMessageIdOnMiddleOfTopic"; |
| final int numOfMessage = 100; |
| final int halfMessages = numOfMessage / 2; |
| |
| Producer<byte[]> producer = pulsarClient.newProducer() |
| .topic(topicName).create(); |
| |
| for (int i = 0; i < numOfMessage; i++) { |
| producer.send(String.format("msg num %d", i).getBytes()); |
| } |
| |
| Reader<byte[]> reader = pulsarClient.newReader().topic(topicName) |
| .startMessageId(MessageId.earliest).create(); |
| |
| assertTrue(reader.hasMessageAvailable()); |
| |
| // Read all messages the first time |
| MessageId midmessageToSeek = null; |
| Set<String> messageSetA = Sets.newHashSet(); |
| for (int i = 0; i < numOfMessage; i++) { |
| Message<byte[]> message = reader.readNext(); |
| String receivedMessage = new String(message.getData()); |
| String expectedMessage = String.format("msg num %d", i); |
| testMessageOrderAndDuplicates(messageSetA, receivedMessage, expectedMessage); |
| |
| if (i == halfMessages) { |
| midmessageToSeek = message.getMessageId(); |
| } |
| } |
| |
| assertFalse(reader.hasMessageAvailable()); |
| |
| // Perform cursor reset by MessageId to half of the topic |
| reader.seek(midmessageToSeek); |
| |
| // Read all halved messages after seek() |
| Set<String> messageSetB = Sets.newHashSet(); |
| for (int i = halfMessages + 1; i < numOfMessage; i++) { |
| Message<byte[]> message = reader.readNext(); |
| String receivedMessage = new String(message.getData()); |
| String expectedMessage = String.format("msg num %d", i); |
| testMessageOrderAndDuplicates(messageSetB, receivedMessage, expectedMessage); |
| } |
| |
| // Reader should be finished |
| assertTrue(reader.isConnected()); |
| assertFalse(reader.hasMessageAvailable()); |
| assertEquals(((ReaderImpl) reader).getConsumer().numMessagesInQueue(), 0); |
| |
| reader.close(); |
| producer.close(); |
| } |
| |
| @Test |
| public void testReaderIsAbleToSeekWithTimeOnMiddleOfTopic() throws Exception { |
| final String topicName = "persistent://my-property/my-ns/ReaderIsAbleToSeekWithTimeOnMiddleOfTopic"; |
| final int numOfMessage = 10; |
| final int halfMessages = numOfMessage / 2; |
| |
| Producer<byte[]> producer = pulsarClient.newProducer() |
| .topic(topicName).create(); |
| |
| long l = System.currentTimeMillis(); |
| for (int i = 0; i < numOfMessage; i++) { |
| producer.send(String.format("msg num %d", i).getBytes()); |
| Thread.sleep(100); |
| } |
| |
| Reader<byte[]> reader = pulsarClient.newReader().topic(topicName) |
| .startMessageId(MessageId.earliest).create(); |
| |
| int plusTime = (halfMessages + 1) * 100; |
| reader.seek(l + plusTime); |
| |
| Set<String> messageSet = Sets.newHashSet(); |
| for (int i = halfMessages + 1; i < numOfMessage; i++) { |
| Message<byte[]> message = reader.readNext(); |
| String receivedMessage = new String(message.getData()); |
| String expectedMessage = String.format("msg num %d", i); |
| testMessageOrderAndDuplicates(messageSet, receivedMessage, expectedMessage); |
| } |
| |
| reader.close(); |
| producer.close(); |
| } |
| |
| @Test |
| public void testMultiReaderIsAbleToSeekWithTimeOnMiddleOfTopic() throws Exception { |
| final String topicName = "persistent://my-property/my-ns/testMultiReaderIsAbleToSeekWithTimeOnMiddleOfTopic" + System.currentTimeMillis(); |
| final int numOfMessage = 10; |
| final int halfMessages = numOfMessage / 2; |
| admin.topics().createPartitionedTopic(topicName, 3); |
| Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName).create(); |
| long halfTime = 0; |
| for (int i = 0; i < numOfMessage; i++) { |
| if (i == numOfMessage / 2) { |
| halfTime = System.currentTimeMillis(); |
| } |
| producer.send(String.format("msg num %d", i).getBytes()); |
| } |
| Assert.assertTrue(halfTime != 0); |
| Reader<byte[]> reader = pulsarClient.newReader().topic(topicName).startMessageId(MessageId.earliest).create(); |
| |
| reader.seek(halfTime); |
| Set<String> messageSet = Sets.newHashSet(); |
| for (int i = halfMessages + 1; i < numOfMessage; i++) { |
| Message<byte[]> message = reader.readNext(10, TimeUnit.SECONDS); |
| String receivedMessage = new String(message.getData()); |
| Assert.assertTrue(messageSet.add(receivedMessage), "Received duplicate message " + receivedMessage); |
| } |
| reader.close(); |
| producer.close(); |
| } |
| |
| @Test(dataProvider = "variationsForExpectedPos") |
| public void testReaderStartMessageIdAtExpectedPos(boolean batching, boolean startInclusive, int numOfMessages) |
| throws Exception { |
| final String topicName = "persistent://my-property/my-ns/ReaderStartMessageIdAtExpectedPos"; |
| final int resetIndex = new Random().nextInt(numOfMessages); // Choose some random index to reset |
| final int firstMessage = startInclusive ? resetIndex : resetIndex + 1; // First message of reset |
| |
| Producer<byte[]> producer = pulsarClient.newProducer() |
| .topic(topicName) |
| .enableBatching(batching) |
| .create(); |
| |
| CountDownLatch latch = new CountDownLatch(numOfMessages); |
| |
| final AtomicReference<MessageId> resetPos = new AtomicReference<>(); |
| |
| for (int i = 0; i < numOfMessages; i++) { |
| |
| final int j = i; |
| |
| producer.sendAsync(String.format("msg num %d", i).getBytes()) |
| .thenCompose(messageId -> FutureUtils.value(Pair.of(j, messageId))) |
| .whenComplete((p, e) -> { |
| if (e != null) { |
| fail("send msg failed due to " + e.getMessage()); |
| } else { |
| if (p.getLeft() == resetIndex) { |
| resetPos.set(p.getRight()); |
| } |
| } |
| latch.countDown(); |
| }); |
| } |
| |
| latch.await(); |
| |
| ReaderBuilder<byte[]> readerBuilder = pulsarClient.newReader() |
| .topic(topicName) |
| .startMessageId(resetPos.get()); |
| |
| if (startInclusive) { |
| readerBuilder.startMessageIdInclusive(); |
| } |
| |
| Reader<byte[]> reader = readerBuilder.create(); |
| Set<String> messageSet = Sets.newHashSet(); |
| for (int i = firstMessage; i < numOfMessages; i++) { |
| Message<byte[]> message = reader.readNext(); |
| String receivedMessage = new String(message.getData()); |
| String expectedMessage = String.format("msg num %d", i); |
| testMessageOrderAndDuplicates(messageSet, receivedMessage, expectedMessage); |
| } |
| |
| assertTrue(reader.isConnected()); |
| assertEquals(((ReaderImpl) reader).getConsumer().numMessagesInQueue(), 0); |
| |
| // Processed messages should be the number of messages in the range: [FirstResetMessage..TotalNumOfMessages] |
| assertEquals(messageSet.size(), numOfMessages - firstMessage); |
| |
| reader.close(); |
| producer.close(); |
| } |
| |
| @Test |
| public void testReaderBuilderConcurrentCreate() throws Exception { |
| String topicName = "persistent://my-property/my-ns/testReaderBuilderConcurrentCreate_"; |
| int numTopic = 30; |
| ReaderBuilder<byte[]> builder = pulsarClient.newReader().startMessageId(MessageId.earliest); |
| |
| List<CompletableFuture<Reader<byte[]>>> readers = Lists.newArrayListWithExpectedSize(numTopic); |
| List<Producer<byte[]>> producers = Lists.newArrayListWithExpectedSize(numTopic); |
| // create producer firstly |
| for (int i = 0; i < numTopic; i++) { |
| producers.add(pulsarClient.newProducer() |
| .topic(topicName + i) |
| .create()); |
| } |
| |
| // create reader concurrently |
| for (int i = 0; i < numTopic; i++) { |
| readers.add(builder.clone().topic(topicName + i).createAsync()); |
| } |
| |
| // verify readers config are different for topic name. |
| for (int i = 0; i < numTopic; i++) { |
| assertEquals(readers.get(i).get().getTopic(), topicName + i); |
| readers.get(i).get().close(); |
| producers.get(i).close(); |
| } |
| } |
| |
| @Test(timeOut = 10000) |
| public void testMultiReaderBuilderConcurrentCreate() throws Exception { |
| String topicName = "persistent://my-property/my-ns/testMultiReaderBuilderConcurrentCreate_"; |
| int numTopic = 30; |
| ReaderBuilder<byte[]> builder = pulsarClient.newReader().startMessageId(MessageId.earliest); |
| |
| List<CompletableFuture<Reader<byte[]>>> readers = Lists.newArrayListWithExpectedSize(numTopic); |
| List<Producer<byte[]>> producers = Lists.newArrayListWithExpectedSize(numTopic); |
| // create producer firstly |
| for (int i = 0; i < numTopic; i++) { |
| admin.topics().createPartitionedTopic(topicName + i, 3); |
| producers.add(pulsarClient.newProducer() |
| .topic(topicName + i) |
| .create()); |
| } |
| |
| // create reader concurrently |
| for (int i = 0; i < numTopic; i++) { |
| readers.add(builder.clone().topic(topicName + i).createAsync()); |
| } |
| |
| // verify readers config are different for topic name. |
| for (int i = 0; i < numTopic; i++) { |
| assertTrue(readers.get(i).get().getTopic().startsWith(MultiTopicsConsumerImpl.DUMMY_TOPIC_NAME_PREFIX)); |
| readers.get(i).get().close(); |
| producers.get(i).close(); |
| } |
| } |
| |
| @Test |
| public void testReaderStartInMiddleOfBatch() throws Exception { |
| final String topicName = "persistent://my-property/my-ns/ReaderStartInMiddleOfBatch"; |
| final int numOfMessage = 100; |
| |
| Producer<byte[]> producer = pulsarClient.newProducer() |
| .topic(topicName) |
| .enableBatching(true) |
| .batchingMaxMessages(10) |
| .create(); |
| |
| CountDownLatch latch = new CountDownLatch(numOfMessage); |
| |
| List<MessageId> allIds = Collections.synchronizedList(new ArrayList<>()); |
| |
| for (int i = 0; i < numOfMessage; i++) { |
| producer.sendAsync(String.format("msg num %d", i).getBytes()).whenComplete((mid, e) -> { |
| if (e != null) { |
| fail(); |
| } else { |
| allIds.add(mid); |
| } |
| latch.countDown(); |
| }); |
| } |
| |
| latch.await(); |
| |
| for (MessageId id : allIds) { |
| Reader<byte[]> reader = pulsarClient.newReader().topic(topicName) |
| .startMessageId(id).startMessageIdInclusive().create(); |
| MessageId idGot = reader.readNext().getMessageId(); |
| assertEquals(idGot, id); |
| reader.close(); |
| } |
| |
| producer.close(); |
| } |
| |
| @Test |
| public void testHasMessageAvailableOnEmptyTopic() throws Exception { |
| String topic = newTopicName(); |
| |
| @Cleanup |
| Reader<String> r1 = pulsarClient.newReader(Schema.STRING) |
| .topic(topic) |
| .startMessageId(MessageId.earliest) |
| .create(); |
| |
| @Cleanup |
| Reader<String> r2 = pulsarClient.newReader(Schema.STRING) |
| .topic(topic) |
| .startMessageId(MessageId.latest) |
| .create(); |
| |
| @Cleanup |
| Reader<String> r2Inclusive = pulsarClient.newReader(Schema.STRING) |
| .topic(topic) |
| .startMessageId(MessageId.latest) |
| .startMessageIdInclusive() |
| .create(); |
| |
| // no data write, should return false |
| assertFalse(r1.hasMessageAvailable()); |
| assertFalse(r2.hasMessageAvailable()); |
| assertFalse(r2Inclusive.hasMessageAvailable()); |
| |
| @Cleanup |
| Producer<String> producer = pulsarClient.newProducer(Schema.STRING) |
| .topic(topic) |
| .create(); |
| |
| producer.send("hello-1"); |
| assertTrue(r1.hasMessageAvailable()); |
| assertTrue(r2.hasMessageAvailable()); |
| assertTrue(r2Inclusive.hasMessageAvailable()); |
| |
| @Cleanup |
| Reader<String> r3 = pulsarClient.newReader(Schema.STRING) |
| .topic(topic) |
| .startMessageId(MessageId.latest) |
| .create(); |
| |
| |
| assertFalse(r3.hasMessageAvailable()); |
| |
| producer.send("hello-2"); |
| |
| assertTrue(r1.hasMessageAvailable()); |
| assertTrue(r2.hasMessageAvailable()); |
| assertTrue(r2Inclusive.hasMessageAvailable()); |
| assertTrue(r3.hasMessageAvailable()); |
| } |
| } |
