@Test(invocationTimeOut = 1000) public void testNotifyPendingReceivedCallback_CompleteWithExceptionWhenMessageIsNull() { CompletableFuture<Message<ConsumerImpl>> receiveFuture = new CompletableFuture<>(); consumer.pendingReceives.add(receiveFuture); consumer.notifyPendingReceivedCallback(null, null); try { receiveFuture.join(); } catch (CompletionException e) { Assert.assertEquals("received message can't be null", e.getCause().getMessage()); } Assert.assertTrue(receiveFuture.isCompletedExceptionally()); }
void notifyPendingReceivedCallback(final Message<T> message, Exception exception) { if (pendingReceives.isEmpty()) { return; } final CompletableFuture<Message<T>> receivedFuture = pendingReceives.poll(); if (receivedFuture == null) { return; } if (exception != null) { listenerExecutor.execute(() -> receivedFuture.completeExceptionally(exception)); return; } if (message == null) { IllegalStateException e = new IllegalStateException("received message can't be null"); listenerExecutor.execute(() -> receivedFuture.completeExceptionally(e)); return; } if (conf.getReceiverQueueSize() == 0) { trackMessage(message); interceptAndComplete(message, receivedFuture); return; } messageProcessed(message); interceptAndComplete(message, receivedFuture); }
ConsumerImpl extends ConsumerBase<T> implements ConnectionHandler.Connection { void notifyPendingReceivedCallback(final Message<T> message, Exception exception) { if (pendingReceives.isEmpty()) { return; } final CompletableFuture<Message<T>> receivedFuture = pendingReceives.poll(); if (receivedFuture == null) { return; } if (exception != null) { listenerExecutor.execute(() -> receivedFuture.completeExceptionally(exception)); return; } if (message == null) { IllegalStateException e = new IllegalStateException("received message can't be null"); listenerExecutor.execute(() -> receivedFuture.completeExceptionally(e)); return; } if (conf.getReceiverQueueSize() == 0) { trackMessage(message); interceptAndComplete(message, receivedFuture); return; } messageProcessed(message); interceptAndComplete(message, receivedFuture); } }
ConsumerImpl extends ConsumerBase<T> implements ConnectionHandler.Connection { void notifyPendingReceivedCallback(final Message<T> message, Exception exception) { if (pendingReceives.isEmpty()) { return; } final CompletableFuture<Message<T>> receivedFuture = pendingReceives.poll(); if (receivedFuture == null) { return; } if (exception != null) { listenerExecutor.execute(() -> receivedFuture.completeExceptionally(exception)); return; } if (message == null) { IllegalStateException e = new IllegalStateException("received message can't be null"); listenerExecutor.execute(() -> receivedFuture.completeExceptionally(e)); return; } if (conf.getReceiverQueueSize() == 0) { trackMessage(message); interceptAndComplete(message, receivedFuture); return; } messageProcessed(message); interceptAndComplete(message, receivedFuture); } protected  ConsumerImpl(PulsarClientImpl client, String topic, ConsumerConfigurationData<T> conf,
            ExecutorService listenerExecutor, int partitionIndex, boolean hasParentConsumer,
            CompletableFuture<Consumer<T>> subscribeFuture, MessageId startMessageId,
            long startMessageRollbackDurationInSec, Schema<T> schema, ConsumerInterceptors<T> interceptors,
            boolean createTopicIfDoesNotExist); }
ConsumerImpl extends ConsumerBase<T> implements ConnectionHandler.Connection { void notifyPendingReceivedCallback(final Message<T> message, Exception exception) { if (pendingReceives.isEmpty()) { return; } final CompletableFuture<Message<T>> receivedFuture = pendingReceives.poll(); if (receivedFuture == null) { return; } if (exception != null) { listenerExecutor.execute(() -> receivedFuture.completeExceptionally(exception)); return; } if (message == null) { IllegalStateException e = new IllegalStateException("received message can't be null"); listenerExecutor.execute(() -> receivedFuture.completeExceptionally(e)); return; } if (conf.getReceiverQueueSize() == 0) { trackMessage(message); interceptAndComplete(message, receivedFuture); return; } messageProcessed(message); interceptAndComplete(message, receivedFuture); } protected  ConsumerImpl(PulsarClientImpl client, String topic, ConsumerConfigurationData<T> conf,
            ExecutorService listenerExecutor, int partitionIndex, boolean hasParentConsumer,
            CompletableFuture<Consumer<T>> subscribeFuture, MessageId startMessageId,
            long startMessageRollbackDurationInSec, Schema<T> schema, ConsumerInterceptors<T> interceptors,
            boolean createTopicIfDoesNotExist); ConnectionHandler getConnectionHandler(); UnAckedMessageTracker getUnAckedMessageTracker(); @Override CompletableFuture<Void> unsubscribeAsync(); @Override void negativeAcknowledge(MessageId messageId); @Override void connectionOpened(final ClientCnx cnx); @Override void connectionFailed(PulsarClientException exception); @Override CompletableFuture<Void> closeAsync(); @Override void pause(); @Override void resume(); @Override boolean isConnected(); @Override int getAvailablePermits(); @Override int numMessagesInQueue(); @Override void redeliverUnacknowledgedMessages(); @Override void redeliverUnacknowledgedMessages(Set<MessageId> messageIds); @Override void seek(MessageId messageId); @Override void seek(long timestamp); @Override CompletableFuture<Void> seekAsync(long timestamp); @Override CompletableFuture<Void> seekAsync(MessageId messageId); boolean hasMessageAvailable(); CompletableFuture<Boolean> hasMessageAvailableAsync(); @Override CompletableFuture<MessageId> getLastMessageIdAsync(); @Override ConsumerStats getStats(); @Override boolean hasReachedEndOfTopic(); @Override int hashCode(); @Override boolean equals(Object o); @VisibleForTesting ClientCnx getClientCnx(); String getTopicNameWithoutPartition(); }
ConsumerImpl extends ConsumerBase<T> implements ConnectionHandler.Connection { void notifyPendingReceivedCallback(final Message<T> message, Exception exception) { if (pendingReceives.isEmpty()) { return; } final CompletableFuture<Message<T>> receivedFuture = pendingReceives.poll(); if (receivedFuture == null) { return; } if (exception != null) { listenerExecutor.execute(() -> receivedFuture.completeExceptionally(exception)); return; } if (message == null) { IllegalStateException e = new IllegalStateException("received message can't be null"); listenerExecutor.execute(() -> receivedFuture.completeExceptionally(e)); return; } if (conf.getReceiverQueueSize() == 0) { trackMessage(message); interceptAndComplete(message, receivedFuture); return; } messageProcessed(message); interceptAndComplete(message, receivedFuture); } protected  ConsumerImpl(PulsarClientImpl client, String topic, ConsumerConfigurationData<T> conf,
            ExecutorService listenerExecutor, int partitionIndex, boolean hasParentConsumer,
            CompletableFuture<Consumer<T>> subscribeFuture, MessageId startMessageId,
            long startMessageRollbackDurationInSec, Schema<T> schema, ConsumerInterceptors<T> interceptors,
            boolean createTopicIfDoesNotExist); ConnectionHandler getConnectionHandler(); UnAckedMessageTracker getUnAckedMessageTracker(); @Override CompletableFuture<Void> unsubscribeAsync(); @Override void negativeAcknowledge(MessageId messageId); @Override void connectionOpened(final ClientCnx cnx); @Override void connectionFailed(PulsarClientException exception); @Override CompletableFuture<Void> closeAsync(); @Override void pause(); @Override void resume(); @Override boolean isConnected(); @Override int getAvailablePermits(); @Override int numMessagesInQueue(); @Override void redeliverUnacknowledgedMessages(); @Override void redeliverUnacknowledgedMessages(Set<MessageId> messageIds); @Override void seek(MessageId messageId); @Override void seek(long timestamp); @Override CompletableFuture<Void> seekAsync(long timestamp); @Override CompletableFuture<Void> seekAsync(MessageId messageId); boolean hasMessageAvailable(); CompletableFuture<Boolean> hasMessageAvailableAsync(); @Override CompletableFuture<MessageId> getLastMessageIdAsync(); @Override ConsumerStats getStats(); @Override boolean hasReachedEndOfTopic(); @Override int hashCode(); @Override boolean equals(Object o); @VisibleForTesting ClientCnx getClientCnx(); String getTopicNameWithoutPartition();  }