How to handle events from Azure Eventhub on multiple partitions one by one by an Azure function app locally

Question

While debugging a Python azure function app locally, that processes events from an Azure Eventhub with 32 paritions I would like to know whether it is possible to handle the events one by one. It seems that whatever setting I set to limit the number of events that arrive are always equal to the number of partitions, in this case 32.

My host.json:

{
  "version": "2.0",
  "extensionBundle": {
    "id": "Microsoft.Azure.Functions.ExtensionBundle",
    "version": "[4.*, 5.0.0)"
  },
  "extensions": {
    "eventHubs": {
      "batchCheckpointFrequency": 1,
      "eventProcessorOptions": {
        "maxBatchSize": 1,
        "prefetchCount": 1,
        "maxConcurrentCalls": 1
      },
      "initialOffsetOptions": {
        "type": "fromEnd"
      }
    }
  }
}

The function app code:

@app.event_hub_message_trigger(arg_name="azeventhub",
                               event_hub_name="eventhubname",
                               connection="EventHubConnectionString",
                               cardinality=func.Cardinality.ONE,
                               consumer_group="consumer_group_x")
async def event_processor(azeventhub: func.EventHubEvent):
    event = azeventhub.get_body().decode('utf-8')
    logging.info("Received from EventHub: %s", event)
    try:
        json_event = json.loads(event)
        await process_event(json_event)

    except Exception as e:
        logging.error("Error while processing event: %s", e)
    time.sleep(10) # it does not wait after each event

In the above code the sleep is only executed after it has processed 32 events, so I assume it sleeps 32 times 10 seconds at the same time. The same behaviour happens when I make the code synchronous.

Is this not possible by design or are there settings that really can force this?

Answer 1

To Process event one by one from Batch of events. Set cardinality to one as mentioned in this MS Doc.

I have used service bus client to receive processe events as message one by one after triggered by EventHub Trigger.

This code worked for me function_app.py:

import azure.functions as func
import logging
import json
import os
from azure.servicebus import ServiceBusClient, ServiceBusMessage

app = func.FunctionApp()

conn = os.getenv("ServiceBus_String")
queue = "queue_event"

def send_message_to_service_bus_queue(message):
    servicebus_client = ServiceBusClient.from_connection_string(conn)
    sender = servicebus_client.get_queue_sender(queue_name=queue)
    sender.send_messages(ServiceBusMessage(message))

@app.event_hub_message_trigger(arg_name="azeventhub", event_hub_name="myeventhub",
                               connection="EventHub_string",cardinality=func.Cardinality.ONE)
def main(azeventhub: func.EventHubEvent):

    events = json.loads(azeventhub.get_body().decode("utf-8"))
    for event in events:
        send_message_to_service_bus_queue(json.dumps(event))
        logging.info("processed message: %s",event)

local.settings.json:

{
  "IsEncrypted": false,
  "Values": {
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "AzureWebJobsFeatureFlags": "EnableWorkerIndexing",
    "EventHub_string": "Endpoint=sb://azevent.servicebus.windows.net/;SharedAccessKeyName=xxxxxxxxxxxxx;SharedAccessKey=xxxxxx",
    "ServiceBus_String":"Endpoint=sb://event-servicebus.servicebus.windows.net/;SharedAccessKeyName=xxxxxxxxxxx;SharedAccessKey=xxxxxxxxxxxxxx"
  }
}

INPUT:

[
    {
        "Name": "Vivek",
        "Contact": "0000-000-000"
    },
    {
        "Name": "Vaibhav",
        "Contact": "1111-111-111"
    },
    {
        "Name": "Shandilya",
        "Contact": "2222-222-222"
    },
    {
        "Name": "Arjun",
        "Contact": "3333-333-333"
    },
    {
        "Name": "Karam",
        "Contact": "4444-444-444"
    }
]

OUTPUT:

events processed one by one and sent to service bus one by one as a single message:

Functions:

        main: eventHubTrigger

For detailed output, run func with --verbose flag.
[2024-02-14T12:00:12.883Z] Host lock lease acquired by instance ID '000000000000000000000000116B5F66'.
[2024-02-14T12:00:17.956Z] Executing 'Functions.main' (Reason='(null)', Id=be44d089-ef21-4028-93e5-1a8cf0d83b61)
[2024-02-14T12:00:17.964Z] Trigger Details: PartionId: 0, Offset: 23184, EnqueueTimeUtc: 2024-02-14T12:00:17.2660000+00:00, SequenceNumber: 39, Count: 1
[2024-02-14T12:00:20.760Z] processed message: {'Name': 'Vivek', 'Contact': '0000-000-000'}
[2024-02-14T12:00:23.828Z] processed message: {'Name': 'Vaibhav', 'Contact': '1111-111-111'}
[2024-02-14T12:00:26.299Z] processed message: {'Name': 'Shandilya', 'Contact': '2222-222-222'}
[2024-02-14T12:00:28.832Z] processed message: {'Name': 'Arjun', 'Contact': '3333-333-333'}
[2024-02-14T12:00:31.448Z] processed message: {'Name': 'Karam', 'Contact': '4444-444-444'}
[2024-02-14T12:00:31.516Z] Executed 'Functions.main' (Succeeded, Id=be44d089-ef21-4028-93e5-1a8cf0d83b61, Duration=13623ms)