Azure function eventhub batch output bindings

Question

I have a NodeJS azure function which receives a batch of event from EventHubA. It does some processing on the individual events which takes around ~3 seconds and send a new event to EventHubB.

I want to sent data to EventHubB as soon as one of the event in batch is processed; and not wait for the whole batch to be processed THEN send all event outputs at once.

I've tried example at EventHub Output binding docs but this seems to be sending output events at the end of function.

I was expecting it to push events to EventhubB as soon as they are pushed to the list.

Is there anyway to send output events as soon as they are processed without waiting for the whole batch to complete? The purpose is to reduce latency of output events.

Here is the index.js

 module.exports = async function (context, events) {
  context.bindings.outputEventHubEvent = [];
  for (let i = 0; i < events.length; i++) {
    let event = JSON.parse(events[i]);
    let output = await someFunctionWhichTakes5Seconds(event);
    context.bindings.outputEventHubEvent.push(output);
   }
  context.done();
};

function.json

{
"bindings": [
    {
        "type": "eventHubTrigger",
        "name": "eventhubA",
        "direction": "in",
        "eventHubName": "EventHubA",
        "connection": "ConnectionStringA",
        "cardinality": "many",
        "consumerGroup": "Sample",
        "dataType": "string",
        "maxBatchSize": 512
    },
    {
        "type": "eventHub",
        "name": "outputEventHubEvent",
        "eventHubName": "EventHubB",
        "connection": "ConnectionStringB",
        "direction": "out"
    }
]
}

I want to send the output event to eventHub as soon as it is generated, not after whole batch is processed.

Answer 1

I am able to send processed message one by one to second EventHub using this code.

For reference used MS Doc.

I used default EventHub trigger sample code and made some changes as per the requirement in code.

index.js:

module.exports = async function (context, eventHubMessages) {
    context.log(`JavaScript eventhub trigger function called for message array ${eventHubMessages}`);

    context.bindings.outputEventHubMessage=[];
    
    eventHubMessages.forEach((message, index) => {
        context.log(`Processed message ${message}`);
        
        context.bindings.outputEventHubMessage = message;

        context.log(`send processed message ${message}`);
        
    });
};

function.json:

{
  "bindings": [
    {
      "type": "eventHubTrigger",
      "name": "eventHubMessages",
      "direction": "in",
      "eventHubName": "batch",
      "connection": "ConnectionStringA",
      "cardinality": "many",
      "consumerGroup": "$Default",
      "dataType": "string"
    },
    {
      "type": "eventHub",
      "name": "outputEventHubMessage",
      "eventHubName": "batch2",
      "connection": "ConnectionStringB",
      "direction": "out"
  }
  ]
}

Output: