@Test(timeOut = 5000) public void testParallelSubscribeAsync() throws Exception { String topicName = "parallel-subscribe-async-topic"; String subscriptionName = "parallel-subscribe-async-subscription"; String serviceUrl = "pulsar: Schema<byte[]> schema = Schema.BYTES; ExecutorService listenerExecutor = mock(ExecutorService.class); ClientConfigurationData conf = new ClientConfigurationData(); conf.setServiceUrl(serviceUrl); ConsumerConfigurationData<byte[]> consumerConfData = new ConsumerConfigurationData<>(); consumerConfData.setSubscriptionName(subscriptionName); int completionDelayMillis = 100; PulsarClientImpl client = setUpPulsarClientMock(schema, completionDelayMillis); MultiTopicsConsumerImpl<byte[]> impl = new MultiTopicsConsumerImpl<>(client, consumerConfData, listenerExecutor, new CompletableFuture<>(), schema, null, true); CompletableFuture<Void> firstInvocation = impl.subscribeAsync(topicName, true); Thread.sleep(5); CompletableFuture<Void> secondInvocation = impl.subscribeAsync(topicName, true); firstInvocation.get(); Throwable t = expectThrows(ExecutionException.class, secondInvocation::get); Throwable cause = t.getCause(); assertEquals(cause.getClass(), PulsarClientException.class); assertTrue(cause.getMessage().endsWith("Topic is already being subscribed for in other thread.")); }
public CompletableFuture<Void> subscribeAsync(String topicName, boolean createTopicIfDoesNotExist) { TopicName topicNameInstance = getTopicName(topicName); if (topicNameInstance == null) { return FutureUtil.failedFuture( new PulsarClientException.AlreadyClosedException("Topic name not valid")); } String fullTopicName = topicNameInstance.toString(); if (topics.containsKey(fullTopicName) || topics.containsKey(topicNameInstance.getPartitionedTopicName())) { return FutureUtil.failedFuture( new PulsarClientException.AlreadyClosedException("Already subscribed to " + topicName)); } if (getState() == State.Closing || getState() == State.Closed) { return FutureUtil.failedFuture( new PulsarClientException.AlreadyClosedException("Topics Consumer was already closed")); } CompletableFuture<Void> subscribeResult = new CompletableFuture<>(); client.getPartitionedTopicMetadata(topicName) .thenAccept(metadata -> subscribeTopicPartitions(subscribeResult, fullTopicName, metadata.partitions, createTopicIfDoesNotExist)) .exceptionally(ex1 -> { log.warn("[{}] Failed to get partitioned topic metadata: {}", fullTopicName, ex1.getMessage()); subscribeResult.completeExceptionally(ex1); return null; }); return subscribeResult; }
MultiTopicsConsumerImpl extends ConsumerBase<T> { public CompletableFuture<Void> subscribeAsync(String topicName, boolean createTopicIfDoesNotExist) { TopicName topicNameInstance = getTopicName(topicName); if (topicNameInstance == null) { return FutureUtil.failedFuture( new PulsarClientException.AlreadyClosedException("Topic name not valid")); } String fullTopicName = topicNameInstance.toString(); if (topics.containsKey(fullTopicName) || topics.containsKey(topicNameInstance.getPartitionedTopicName())) { return FutureUtil.failedFuture( new PulsarClientException.AlreadyClosedException("Already subscribed to " + topicName)); } if (getState() == State.Closing || getState() == State.Closed) { return FutureUtil.failedFuture( new PulsarClientException.AlreadyClosedException("Topics Consumer was already closed")); } CompletableFuture<Void> subscribeResult = new CompletableFuture<>(); client.getPartitionedTopicMetadata(topicName) .thenAccept(metadata -> subscribeTopicPartitions(subscribeResult, fullTopicName, metadata.partitions, createTopicIfDoesNotExist)) .exceptionally(ex1 -> { log.warn("[{}] Failed to get partitioned topic metadata: {}", fullTopicName, ex1.getMessage()); subscribeResult.completeExceptionally(ex1); return null; }); return subscribeResult; } }
MultiTopicsConsumerImpl extends ConsumerBase<T> { public CompletableFuture<Void> subscribeAsync(String topicName, boolean createTopicIfDoesNotExist) { TopicName topicNameInstance = getTopicName(topicName); if (topicNameInstance == null) { return FutureUtil.failedFuture( new PulsarClientException.AlreadyClosedException("Topic name not valid")); } String fullTopicName = topicNameInstance.toString(); if (topics.containsKey(fullTopicName) || topics.containsKey(topicNameInstance.getPartitionedTopicName())) { return FutureUtil.failedFuture( new PulsarClientException.AlreadyClosedException("Already subscribed to " + topicName)); } if (getState() == State.Closing || getState() == State.Closed) { return FutureUtil.failedFuture( new PulsarClientException.AlreadyClosedException("Topics Consumer was already closed")); } CompletableFuture<Void> subscribeResult = new CompletableFuture<>(); client.getPartitionedTopicMetadata(topicName) .thenAccept(metadata -> subscribeTopicPartitions(subscribeResult, fullTopicName, metadata.partitions, createTopicIfDoesNotExist)) .exceptionally(ex1 -> { log.warn("[{}] Failed to get partitioned topic metadata: {}", fullTopicName, ex1.getMessage()); subscribeResult.completeExceptionally(ex1); return null; }); return subscribeResult; } MultiTopicsConsumerImpl(PulsarClientImpl client, ConsumerConfigurationData<T> conf,
            ExecutorService listenerExecutor, CompletableFuture<Consumer<T>> subscribeFuture, Schema<T> schema,
            ConsumerInterceptors<T> interceptors, boolean createTopicIfDoesNotExist);   MultiTopicsConsumerImpl(PulsarClientImpl client, String singleTopic, ConsumerConfigurationData<T> conf,
            ExecutorService listenerExecutor, CompletableFuture<Consumer<T>> subscribeFuture, Schema<T> schema,
            ConsumerInterceptors<T> interceptors, boolean createTopicIfDoesNotExist); }
MultiTopicsConsumerImpl extends ConsumerBase<T> { public CompletableFuture<Void> subscribeAsync(String topicName, boolean createTopicIfDoesNotExist) { TopicName topicNameInstance = getTopicName(topicName); if (topicNameInstance == null) { return FutureUtil.failedFuture( new PulsarClientException.AlreadyClosedException("Topic name not valid")); } String fullTopicName = topicNameInstance.toString(); if (topics.containsKey(fullTopicName) || topics.containsKey(topicNameInstance.getPartitionedTopicName())) { return FutureUtil.failedFuture( new PulsarClientException.AlreadyClosedException("Already subscribed to " + topicName)); } if (getState() == State.Closing || getState() == State.Closed) { return FutureUtil.failedFuture( new PulsarClientException.AlreadyClosedException("Topics Consumer was already closed")); } CompletableFuture<Void> subscribeResult = new CompletableFuture<>(); client.getPartitionedTopicMetadata(topicName) .thenAccept(metadata -> subscribeTopicPartitions(subscribeResult, fullTopicName, metadata.partitions, createTopicIfDoesNotExist)) .exceptionally(ex1 -> { log.warn("[{}] Failed to get partitioned topic metadata: {}", fullTopicName, ex1.getMessage()); subscribeResult.completeExceptionally(ex1); return null; }); return subscribeResult; } MultiTopicsConsumerImpl(PulsarClientImpl client, ConsumerConfigurationData<T> conf,
            ExecutorService listenerExecutor, CompletableFuture<Consumer<T>> subscribeFuture, Schema<T> schema,
            ConsumerInterceptors<T> interceptors, boolean createTopicIfDoesNotExist);   MultiTopicsConsumerImpl(PulsarClientImpl client, String singleTopic, ConsumerConfigurationData<T> conf,
            ExecutorService listenerExecutor, CompletableFuture<Consumer<T>> subscribeFuture, Schema<T> schema,
            ConsumerInterceptors<T> interceptors, boolean createTopicIfDoesNotExist); @Override void negativeAcknowledge(MessageId messageId); @Override CompletableFuture<Void> unsubscribeAsync(); @Override CompletableFuture<Void> closeAsync(); @Override boolean isConnected(); @Override void redeliverUnacknowledgedMessages(); @Override void redeliverUnacknowledgedMessages(Set<MessageId> messageIds); @Override void seek(MessageId messageId); @Override void seek(long timestamp); @Override CompletableFuture<Void> seekAsync(MessageId messageId); @Override CompletableFuture<Void> seekAsync(long timestamp); @Override int getAvailablePermits(); @Override boolean hasReachedEndOfTopic(); @Override int numMessagesInQueue(); @Override synchronized ConsumerStats getStats(); UnAckedMessageTracker getUnAckedMessageTracker(); CompletableFuture<Void> subscribeAsync(String topicName, boolean createTopicIfDoesNotExist); static MultiTopicsConsumerImpl<T> createPartitionedConsumer(PulsarClientImpl client,
                                                                           ConsumerConfigurationData<T> conf,
                                                                           ExecutorService listenerExecutor,
                                                                           CompletableFuture<Consumer<T>> subscribeFuture,
                                                                           int numPartitions,
                                                                           Schema<T> schema, ConsumerInterceptors<T> interceptors); CompletableFuture<Void> unsubscribeAsync(String topicName); CompletableFuture<Void> removeConsumerAsync(String topicName); List<String> getTopics(); List<String> getPartitionedTopics(); List<ConsumerImpl<T>> getConsumers(); @Override void pause(); @Override void resume(); @VisibleForTesting Timeout getPartitionsAutoUpdateTimeout(); @Override CompletableFuture<MessageId> getLastMessageIdAsync(); }
MultiTopicsConsumerImpl extends ConsumerBase<T> { public CompletableFuture<Void> subscribeAsync(String topicName, boolean createTopicIfDoesNotExist) { TopicName topicNameInstance = getTopicName(topicName); if (topicNameInstance == null) { return FutureUtil.failedFuture( new PulsarClientException.AlreadyClosedException("Topic name not valid")); } String fullTopicName = topicNameInstance.toString(); if (topics.containsKey(fullTopicName) || topics.containsKey(topicNameInstance.getPartitionedTopicName())) { return FutureUtil.failedFuture( new PulsarClientException.AlreadyClosedException("Already subscribed to " + topicName)); } if (getState() == State.Closing || getState() == State.Closed) { return FutureUtil.failedFuture( new PulsarClientException.AlreadyClosedException("Topics Consumer was already closed")); } CompletableFuture<Void> subscribeResult = new CompletableFuture<>(); client.getPartitionedTopicMetadata(topicName) .thenAccept(metadata -> subscribeTopicPartitions(subscribeResult, fullTopicName, metadata.partitions, createTopicIfDoesNotExist)) .exceptionally(ex1 -> { log.warn("[{}] Failed to get partitioned topic metadata: {}", fullTopicName, ex1.getMessage()); subscribeResult.completeExceptionally(ex1); return null; }); return subscribeResult; } MultiTopicsConsumerImpl(PulsarClientImpl client, ConsumerConfigurationData<T> conf,
            ExecutorService listenerExecutor, CompletableFuture<Consumer<T>> subscribeFuture, Schema<T> schema,
            ConsumerInterceptors<T> interceptors, boolean createTopicIfDoesNotExist);   MultiTopicsConsumerImpl(PulsarClientImpl client, String singleTopic, ConsumerConfigurationData<T> conf,
            ExecutorService listenerExecutor, CompletableFuture<Consumer<T>> subscribeFuture, Schema<T> schema,
            ConsumerInterceptors<T> interceptors, boolean createTopicIfDoesNotExist); @Override void negativeAcknowledge(MessageId messageId); @Override CompletableFuture<Void> unsubscribeAsync(); @Override CompletableFuture<Void> closeAsync(); @Override boolean isConnected(); @Override void redeliverUnacknowledgedMessages(); @Override void redeliverUnacknowledgedMessages(Set<MessageId> messageIds); @Override void seek(MessageId messageId); @Override void seek(long timestamp); @Override CompletableFuture<Void> seekAsync(MessageId messageId); @Override CompletableFuture<Void> seekAsync(long timestamp); @Override int getAvailablePermits(); @Override boolean hasReachedEndOfTopic(); @Override int numMessagesInQueue(); @Override synchronized ConsumerStats getStats(); UnAckedMessageTracker getUnAckedMessageTracker(); CompletableFuture<Void> subscribeAsync(String topicName, boolean createTopicIfDoesNotExist); static MultiTopicsConsumerImpl<T> createPartitionedConsumer(PulsarClientImpl client,
                                                                           ConsumerConfigurationData<T> conf,
                                                                           ExecutorService listenerExecutor,
                                                                           CompletableFuture<Consumer<T>> subscribeFuture,
                                                                           int numPartitions,
                                                                           Schema<T> schema, ConsumerInterceptors<T> interceptors); CompletableFuture<Void> unsubscribeAsync(String topicName); CompletableFuture<Void> removeConsumerAsync(String topicName); List<String> getTopics(); List<String> getPartitionedTopics(); List<ConsumerImpl<T>> getConsumers(); @Override void pause(); @Override void resume(); @VisibleForTesting Timeout getPartitionsAutoUpdateTimeout(); @Override CompletableFuture<MessageId> getLastMessageIdAsync(); static final String DUMMY_TOPIC_NAME_PREFIX; }