Null-proof message ignored listener

The message ignored listener in the consumer coordinator calls the
processed method on ignored messages to "simulate" the processing (e.g.
increment a counter to provide a credit in the middle of the chunk).

The MessageProcessedCallback is the chunk context in this case, which
can be null, depending on the consumer flow strategy. The listener could
trigger a null-pointer exception by calling the processed method on the
context.

This commit adds a null check to avoid the null-pointer exception. A null
chunk context (MessageProcessedCallback) means the custom behavior of
the consumer flow strategy is performed when the chunk starts, so the
call to MessageHandler.Context#processed() is not necessary.

Author: acogoluegnes

src/main/java/com/rabbitmq/stream/impl/ConsumersCoordinator.java

@@ -717,16 +717,18 @@ private ClientSubscriptionsManager(
                 subscriptionTrackers.get(subscriptionId & 0xFF);
             if (subscriptionTracker != null) {
               // message at the beginning of the first chunk is ignored
-              // we "simulate" the processing
-              MessageHandlerContext messageHandlerContext =
-                  new MessageHandlerContext(
-                      offset,
-                      chunkTimestamp,
-                      committedChunkId,
-                      subscriptionTracker.consumer,
-                      (ConsumerFlowStrategy.MessageProcessedCallback) chunkContext);
-              ((ConsumerFlowStrategy.MessageProcessedCallback) chunkContext)
-                  .processed(messageHandlerContext);
+              // we "simulate" the processing if possible
+              if (chunkContext != null) {
+                MessageHandlerContext messageHandlerContext =
+                    new MessageHandlerContext(
+                        offset,
+                        chunkTimestamp,
+                        committedChunkId,
+                        subscriptionTracker.consumer,
+                        (ConsumerFlowStrategy.MessageProcessedCallback) chunkContext);
+                ((ConsumerFlowStrategy.MessageProcessedCallback) chunkContext)
+                    .processed(messageHandlerContext);
+              }
             } else {
               LOGGER.debug(
                   "Could not find stream subscription {} in manager {}, node {} for message ignored listener",