dlg99 commented on a change in pull request #10025:
URL: https://github.com/apache/pulsar/pull/10025#discussion_r616861090
##########
File path:
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
##########
@@ -273,17 +273,34 @@ protected void
failPendingBatchReceives(ConcurrentLinkedQueue<OpBatchReceive<T>>
abstract protected CompletableFuture<Messages<T>>
internalBatchReceiveAsync();
+ private static void validateMessageId(Message<?> message) throws
PulsarClientException {
+ if (message == null) {
+ throw new PulsarClientException.InvalidMessageException("Non-null
message is required");
+ }
+ if (message.getMessageId() == null) {
+ throw new PulsarClientException.InvalidMessageException("Cannot
handle message with null messageId");
+ }
+ }
+
+ private static void validateMessageId(MessageId messageId) throws
PulsarClientException {
+ if (messageId == null) {
+ throw new PulsarClientException.InvalidMessageException("Cannot
handle message with null messageId");
+ }
+ }
+
@Override
public void acknowledge(Message<?> message) throws PulsarClientException {
+ validateMessageId(message);
try {
acknowledge(message.getMessageId());
- } catch (NullPointerException npe) {
- throw new
PulsarClientException.InvalidMessageException(npe.getMessage());
+ } catch (Exception e) {
Review comment:
@eolivelli The method throws PulsarClientException / point of this
catch it to wrap other exception types into the PulsarClientException.
acknowledge() internally gets into acknowledgeAsync().get() which can throw
InterruptedException/ExecutionException.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
