@Test public void testGetStats() throws Exception { String topicName = "test-stats"; ClientConfigurationData conf = new ClientConfigurationData(); conf.setServiceUrl("pulsar: conf.setStatsIntervalSeconds(100); ThreadFactory threadFactory = new DefaultThreadFactory("client-test-stats", Thread.currentThread().isDaemon()); EventLoopGroup eventLoopGroup = EventLoopUtil.newEventLoopGroup(conf.getNumIoThreads(), threadFactory); ExecutorService listenerExecutor = Executors.newSingleThreadScheduledExecutor(threadFactory); PulsarClientImpl clientImpl = new PulsarClientImpl(conf, eventLoopGroup); ConsumerConfigurationData consumerConfData = new ConsumerConfigurationData(); consumerConfData.setTopicNames(Sets.newHashSet(topicName)); assertEquals(Long.parseLong("100"), clientImpl.getConfiguration().getStatsIntervalSeconds()); MultiTopicsConsumerImpl impl = new MultiTopicsConsumerImpl( clientImpl, consumerConfData, listenerExecutor, null, null, null, true); impl.getStats(); }
@Override public synchronized ConsumerStats getStats() { if (stats == null) { return null; } stats.reset(); consumers.values().stream().forEach(consumer -> stats.updateCumulativeStats(consumer.getStats())); return stats; }
MultiTopicsConsumerImpl extends ConsumerBase<T> { @Override public synchronized ConsumerStats getStats() { if (stats == null) { return null; } stats.reset(); consumers.values().stream().forEach(consumer -> stats.updateCumulativeStats(consumer.getStats())); return stats; } }
MultiTopicsConsumerImpl extends ConsumerBase<T> { @Override public synchronized ConsumerStats getStats() { if (stats == null) { return null; } stats.reset(); consumers.values().stream().forEach(consumer -> stats.updateCumulativeStats(consumer.getStats())); return stats; } 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> { @Override public synchronized ConsumerStats getStats() { if (stats == null) { return null; } stats.reset(); consumers.values().stream().forEach(consumer -> stats.updateCumulativeStats(consumer.getStats())); return stats; } 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> { @Override public synchronized ConsumerStats getStats() { if (stats == null) { return null; } stats.reset(); consumers.values().stream().forEach(consumer -> stats.updateCumulativeStats(consumer.getStats())); return stats; } 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; }