Fixes issue where lastEnqueuedEventProperties is null for empty window. (Azure#27363)

* Adding test for PartitionPump

* Adding startPartitionPump tests for PartitionPumpManager.

* Adding test for stopping.

* Update PartitionPump to keep track of last received event properties.

* Update PartitionPumpManager to keep track of lastEnqueuedProperties and use those when a batch is empty.

* Adding test case to ensure enqueued properties updated.

* Updating CHANGELOG

Author: conniey

sdk/eventhubs/azure-messaging-eventhubs/CHANGELOG.md

@@ -9,6 +9,7 @@
 ### Bugs Fixed
 
 - Removed the incorrect lock from `EventDataBatch.tryAdd()` implementation and documented that this API is not thread-safe. ([#25910](https://github.com/Azure/azure-sdk-for-java/issues/25910))
+- Fixed a bug where users get a NullPointerException when getting `LastEnqueuedEventProperties` for an empty window. ([#27121](https://github.com/Azure/azure-sdk-for-java/issues/27121))
 
 ### Other Changes
 

sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/PartitionPump.java

@@ -4,6 +4,7 @@
 package com.azure.messaging.eventhubs;
 
 import com.azure.core.util.logging.ClientLogger;
+import com.azure.messaging.eventhubs.models.LastEnqueuedEventProperties;
 import reactor.core.scheduler.Scheduler;
 
 import static com.azure.messaging.eventhubs.implementation.ClientConstants.PARTITION_ID_KEY;
@@ -15,6 +16,7 @@ class PartitionPump implements AutoCloseable {
     private final String partitionId;
     private final EventHubConsumerAsyncClient client;
     private final Scheduler scheduler;
+    private LastEnqueuedEventProperties lastEnqueuedEventProperties;
     private final ClientLogger logger = new ClientLogger(PartitionPump.class);
 
     /**
@@ -34,6 +36,25 @@ EventHubConsumerAsyncClient getClient() {
         return client;
     }
 
+    /**
+     * Gets the last enqueued event properties.
+     *
+     * @return the last enqueued event properties or null if there has been no events received or {@link
+     *     EventProcessorClientBuilder#trackLastEnqueuedEventProperties(boolean)} is false.
+     */
+    LastEnqueuedEventProperties getLastEnqueuedEventProperties() {
+        return lastEnqueuedEventProperties;
+    }
+
+    /**
+     * Sets the last enqueued event properties seen.
+     *
+     * @param lastEnqueuedEventProperties the last enqueued event properties.
+     */
+    void setLastEnqueuedEventProperties(LastEnqueuedEventProperties lastEnqueuedEventProperties) {
+        this.lastEnqueuedEventProperties = lastEnqueuedEventProperties;
+    }
+
     /**
      * Disposes of the scheduler and the consumer.
      */

sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/PartitionPumpManager.java

@@ -254,8 +254,8 @@ void startPartitionPump(PartitionOwnership claimedOwnership, Checkpoint checkpoi
                 .concatMap(Flux::collectList)
                 .publishOn(scheduler, false, prefetch)
                 .subscribe(partitionEventBatch -> {
-                    processEvents(partitionContext, partitionProcessor,
-                        eventHubConsumer, partitionEventBatch);
+                    processEvents(partitionContext, partitionProcessor, partitionPump, eventHubConsumer,
+                        partitionEventBatch);
                 },
                     /* EventHubConsumer receive() returned an error */
                     ex -> handleError(claimedOwnership, partitionPump, partitionProcessor, ex, partitionContext),
@@ -316,7 +316,8 @@ private void processEvent(PartitionContext partitionContext, PartitionProcessor
     }
 
     private void processEvents(PartitionContext partitionContext, PartitionProcessor partitionProcessor,
-        EventHubConsumerAsyncClient eventHubConsumer, List<PartitionEvent> partitionEventBatch) {
+        PartitionPump partitionPump, EventHubConsumerAsyncClient eventHubConsumer,
+        List<PartitionEvent> partitionEventBatch) {
         try {
             if (batchReceiveMode) {
                 LastEnqueuedEventProperties[] lastEnqueuedEventProperties = new LastEnqueuedEventProperties[1];
@@ -326,15 +327,23 @@ private void processEvents(PartitionContext partitionContext, PartitionProcessor
                         return partitionEvent.getData();
                     })
                     .collect(Collectors.toList());
+
+                // It's possible when using windowTimeout that in the timeframe, there weren't any events received.
+                LastEnqueuedEventProperties enqueuedEventProperties =
+                    updateOrGetLastEnqueuedEventProperties(partitionPump, lastEnqueuedEventProperties[0]);
+
                 EventBatchContext eventBatchContext = new EventBatchContext(partitionContext, eventDataList,
-                    checkpointStore, lastEnqueuedEventProperties[0]);
+                    checkpointStore, enqueuedEventProperties);
+
                 if (logger.canLogAtLevel(LogLevel.VERBOSE)) {
                     logger.atVerbose()
                         .addKeyValue(PARTITION_ID_KEY, partitionContext.getPartitionId())
                         .addKeyValue(ENTITY_PATH_KEY, partitionContext.getEventHubName())
                         .log("Processing event batch.");
                 }
+
                 partitionProcessor.processEventBatch(eventBatchContext);
+
                 if (logger.canLogAtLevel(LogLevel.VERBOSE)) {
                     logger.atVerbose()
                         .addKeyValue(PARTITION_ID_KEY, partitionContext.getPartitionId())
@@ -346,8 +355,14 @@ private void processEvents(PartitionContext partitionContext, PartitionProcessor
                     ? partitionEventBatch.get(0).getData() : null);
                 LastEnqueuedEventProperties lastEnqueuedEventProperties = (partitionEventBatch.size() == 1
                     ? partitionEventBatch.get(0).getLastEnqueuedEventProperties() : null);
+
+                // Get the last value we've seen if the current value from this is null.
+                LastEnqueuedEventProperties enqueuedEventProperties =
+                    updateOrGetLastEnqueuedEventProperties(partitionPump, lastEnqueuedEventProperties);
+
                 EventContext eventContext = new EventContext(partitionContext, eventData, checkpointStore,
-                    lastEnqueuedEventProperties);
+                    enqueuedEventProperties);
+
                 processEvent(partitionContext, partitionProcessor, eventHubConsumer, eventContext);
             }
         } catch (Throwable throwable) {
@@ -449,4 +464,22 @@ private void endProcessTracingSpan(Context processSpanContext, Signal<Void> sign
         }
         tracerProvider.endSpan(processSpanContext, signal);
     }
+
+    /**
+     * Updates the last enqueued event if it was seen and gets the most up-to-date value.
+     *
+     * @param partitionPump The partition pump.
+     * @param last The last enqueued event properties. Could be null if there were no events seen.
+     *
+     * @return Updates the partition pump properties if there is a latest value and returns it.
+     */
+    private LastEnqueuedEventProperties updateOrGetLastEnqueuedEventProperties(PartitionPump partitionPump,
+        LastEnqueuedEventProperties last) {
+
+        if (last != null) {
+            partitionPump.setLastEnqueuedEventProperties(last);
+        }
+
+        return partitionPump.getLastEnqueuedEventProperties();
+    }
 }