Fixing the logic to get unprocessed messages from the azure event hub partition
@@ -113,6 +113,11 @@ func parseAzureEventHubMetadata(metadata, resolvedEnv map[string]string) (*Event
//GetUnprocessedEventCountInPartition gets number of unprocessed events in a given partition
func (scaler *AzureEventHubScaler) GetUnprocessedEventCountInPartition(ctx context.Context, partitionInfo *eventhub.HubPartitionRuntimeInformation) (newEventCount int64, checkpoint azure.Checkpoint, err error) {
+ //if partitionInfo.LastEnqueuedOffset = -1, that means event hub partition is empty
+ if partitionInfo!= nil && partitionInfo.LastEnqueuedOffset == "-1" {
+ return 0, azure.Checkpoint{}, nil
+ }
+
checkpoint, err = azure.GetCheckpointFromBlobStorage(ctx, scaler.metadata.eventHubInfo, partitionInfo.PartitionID)
if err != nil {
// if blob not found return the total partition event count
@@ -127,15 +132,28 @@ func (scaler *AzureEventHubScaler) GetUnprocessedEventCountInPartition(ctx conte
unprocessedEventCountInPartition := int64(0)
- if checkpoint.SequenceNumber != partitionInfo.LastSequenceNumber {
- if partitionInfo.LastSequenceNumber > checkpoint.SequenceNumber {
- unprocessedEventCountInPartition = partitionInfo.LastSequenceNumber - checkpoint.SequenceNumber
+ //If checkpoint.Offset is empty that means no messages has been processed from an event hub partition
+ // And since partitionInfo.LastSequenceNumber = 0 for the very first message hence
+ // total unprocessed message will be partitionInfo.LastSequenceNumber + 1
+ if checkpoint.Offset == "" {
+ unprocessedEventCountInPartition = partitionInfo.LastSequenceNumber + 1
+ return unprocessedEventCountInPartition, checkpoint, nil
+ }
+ if partitionInfo.LastSequenceNumber >= checkpoint.SequenceNumber {
+ unprocessedEventCountInPartition = partitionInfo.LastSequenceNumber - checkpoint.SequenceNumber
return unprocessedEventCountInPartition, checkpoint, nil
}
+ // Partition is a circular buffer, so it is possible that
+ // partitionInfo.LastSequenceNumber < blob checkpoint's SequenceNumber
unprocessedEventCountInPartition = (math.MaxInt64 - partitionInfo.LastSequenceNumber) + checkpoint.SequenceNumber
- }
+
+ // Checkpointing may or may not be always behind partition's LastSequenceNumber.
+ // The partition information read could be stale compared to checkpoint,
+ // especially when load is very small and checkpointing is happening often.
+ // e.g., (9223372036854775807 - 10) + 11 = -9223372036854775808
+ // If unprocessedEventCountInPartition is negative that means there are 0 unprocessed messages in the partition
if unprocessedEventCountInPartition < 0 {
unprocessedEventCountInPartition = 0
}
