Offset in Azure EventHubTrigger Function App

Question

Expectation: Using a function app with EventHubTrigger-CSharp template runs when a new event is comming in.

Implementation: Using the very basic example from https://azure.microsoft.com/en-us/documentation/articles/functions-bindings-event-hubs/

Result: The explained method:

public static void Run(string myEventHubMessage, TraceWriter log)
{
    log.Info($"C# Event Hub trigger function processed a message: {myEventHubMessage}");
}

does deliver the 'first' in the event queue instead of the current one.

Additional:

• I added a consumerGroup (so not reading from $Default)
• According to this blog: http://www.yourcloudbuddy.com/2016/04/azure-functions-and-event-hubs.html 'there is currently no way to configure it to start a particular offset'.

Question: How can I receive the newest event on a hub in my function app?

More info: I need to forward new signals from event hub to an external mqtt api (that works so far). Probably I need another approach?

Antisepticize

You should probably accept an answer here. I have used the approach in my answer below numerous times to make my Azure Function App read data from the END of an eventhub. – Mayorga

Answer 1

@Ramin, I tried to recreate a scenario similar to yours but could not repro the issue. Here are the steps:

1. Setup an Event Hub-triggered Function (using latest version 0.7 as of 10/27/2016)

2. Setup a simple console app sender to my Event Hub that sends 10 messages with time stamps and message ids. Each message is sent 6 minutes apart. Here's the output for my console app that logs the sent messages,

Sent message: Timestamp: 10/27/2016 2:27:06 PM, MessageId: 1
Sent message: Timestamp: 10/27/2016 2:33:07 PM, MessageId: 2
Sent message: Timestamp: 10/27/2016 2:39:07 PM, MessageId: 3
Sent message: Timestamp: 10/27/2016 2:45:07 PM, MessageId: 4
Sent message: Timestamp: 10/27/2016 2:51:08 PM, MessageId: 5
Sent message: Timestamp: 10/27/2016 2:57:08 PM, MessageId: 6
Sent message: Timestamp: 10/27/2016 3:03:08 PM, MessageId: 7
Sent message: Timestamp: 10/27/2016 3:09:08 PM, MessageId: 8
Sent message: Timestamp: 10/27/2016 3:15:08 PM, MessageId: 9
Sent message: Timestamp: 10/27/2016 3:21:08 PM, MessageId: 10
Stopped sending messages.

Here are the log entries for my Function,

2016-10-27T21:27:26.516 Function started (Id=d5ed0e23-2b0e-4e0b-a858-f1e497dcac8c)
2016-10-27T21:27:26.516 C# Event Hub trigger function processed a message: Timestamp: 10/27/2016 2:27:06 PM, MessageId: 1
2016-10-27T21:27:26.516 Function completed (Success, Id=d5ed0e23-2b0e-4e0b-a858-f1e497dcac8c)
2016-10-27T21:33:26.667 Function started (Id=9525a046-86fb-4499-9d4f-b0d0fd0d0829)
2016-10-27T21:33:26.667 C# Event Hub trigger function processed a message: Timestamp: 10/27/2016 2:33:07 PM, MessageId: 2
2016-10-27T21:33:26.667 Function completed (Success, Id=9525a046-86fb-4499-9d4f-b0d0fd0d0829)
2016-10-27T21:39:42.074 Function started (Id=e2d528c9-f1b9-41aa-9c38-669c57c8182a)
2016-10-27T21:39:42.074 C# Event Hub trigger function processed a message: Timestamp: 10/27/2016 2:39:07 PM, MessageId: 3
2016-10-27T21:39:42.074 Function completed (Success, Id=e2d528c9-f1b9-41aa-9c38-669c57c8182a)
2016-10-27T21:45:26.794 Function started (Id=ff5325af-5dab-48fb-95b1-8318fada3c8c)
2016-10-27T21:45:26.794 C# Event Hub trigger function processed a message: Timestamp: 10/27/2016 2:45:07 PM, MessageId: 4
2016-10-27T21:45:26.794 Function completed (Success, Id=ff5325af-5dab-48fb-95b1-8318fada3c8c)
2016-10-27T21:51:26.875 Function started (Id=01d3fbf9-b772-4076-8fbf-783dc16677a7)
2016-10-27T21:51:26.875 C# Event Hub trigger function processed a message: Timestamp: 10/27/2016 2:51:08 PM, MessageId: 5
2016-10-27T21:51:26.875 Function completed (Success, Id=01d3fbf9-b772-4076-8fbf-783dc16677a7)
2016-10-27T21:57:26.989 Function started (Id=443d96e6-af97-460a-9f70-9dbc2eaf2f37)
2016-10-27T21:57:26.989 C# Event Hub trigger function processed a message: Timestamp: 10/27/2016 2:57:08 PM, MessageId: 6
2016-10-27T21:57:26.989 Function completed (Success, Id=443d96e6-af97-460a-9f70-9dbc2eaf2f37)
2016-10-27T22:03:27.088 Function started (Id=9eefe03d-8e78-4119-a453-96655ea01824)
2016-10-27T22:03:27.088 C# Event Hub trigger function processed a message: Timestamp: 10/27/2016 3:03:08 PM, MessageId: 7
2016-10-27T22:03:27.088 Function completed (Success, Id=9eefe03d-8e78-4119-a453-96655ea01824)
2016-10-27T22:10:01.872 Function started (Id=34c9b277-059d-4839-9dce-aeb03afb2871)
2016-10-27T22:10:01.872 C# Event Hub trigger function processed a message: Timestamp: 10/27/2016 3:09:08 PM, MessageId: 8
2016-10-27T22:10:01.872 Function completed (Success, Id=34c9b277-059d-4839-9dce-aeb03afb2871)
2016-10-27T22:15:27.706 Function started (Id=50eda659-c68f-4e61-a942-32160668fd5c)
2016-10-27T22:15:27.706 C# Event Hub trigger function processed a message: Timestamp: 10/27/2016 3:15:08 PM, MessageId: 9
2016-10-27T22:15:27.706 Function completed (Success, Id=50eda659-c68f-4e61-a942-32160668fd5c)
2016-10-27T22:21:52.162 Function started (Id=fa8f8059-013f-42f9-8047-391d4e3fb4a3)
2016-10-27T22:21:52.162 C# Event Hub trigger function processed a message: Timestamp: 10/27/2016 3:21:08 PM, MessageId: 10
2016-10-27T22:21:52.162 Function completed (Success, Id=fa8f8059-013f-42f9-8047-391d4e3fb4a3)

As you can see, all the messages arrived without duplication. Can you confirm that which each invocation of your Function (you can check your Function logs), that it actually ran to completion successfully?

If you see the following message,

Function completed (Success, Id=<some Guid>)

for every invocation of your Function, then the message(s) for that execution should have been check-pointed, and you should not be processing the message(s) again in the next invocation.

However, if you do not see the message or see an error message, then the Function run failed, which will result in the message(s) not being check-pointed. When the Function gets triggered again, it will pick up from the last checkpoint, which could be the first message, if your Function never succeeded in processing that message.

Acetous

After updating to v0.7 it worked. I had also an issue with my dynamic json object (no warning so far), so obviously you were right with the assumption that the function did not complete correctly. Now it works as it should according to my expectations. Thank you for your detailed examination, Ramin. – Antisepticize

Answer 2

There now is a way.

Note: For this to work, you must delete the consumer group data stored in azure-webjobs-eventhub in the storage account you connected to your function app.

---

We now have the ability to use dependency injection in Azure Functions using the Microsoft.Azure.Functions.Extensions nuget package.

This has opened up a way for us to control where our Azure Function App starts to process our event hub by doing something like this:

using System;
using Microsoft.Azure.EventHubs;
using Microsoft.Azure.Functions.Extensions.DependencyInjection;
using Microsoft.Azure.WebJobs.EventHubs;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;

[assembly: FunctionsStartup(typeof(EventHubOffsetTest.Startup))]

namespace EventHubOffsetTest
{
    public class Startup : FunctionsStartup
    {
        public override void Configure(IFunctionsHostBuilder builder)
        {
            builder.Services.PostConfigure<EventHubOptions>(o => o.EventProcessorOptions.InitialOffsetProvider = GetInitialOffset);
        }

        private static EventPosition GetInitialOffset(string partitionId)
        {
            return EventPosition.FromEnd();
        }
    }
}

Mayorga

For anyone using local development and azurite the same idea applies as @Mayorga stated. In your root /azurite folder there is a file azurite_db_blob.json. Remove all blocks that match 'azure-webjobs-eventhub' and restart azurite. It should reset your checkpoint back to zero. This is handy when deleting eventhubs during development. – Martinelli

Answer 3

EventHub listening is based on the EventProcessorHost class (https://msdn.microsoft.com/en-us/library/azure/microsoft.servicebus.messaging.eventprocessorhost.aspx) and has those default semantics. You're correct, you can't control the starting offset. It just starts listening and checkpointing.

We also track additional information at the wiki: https://github.com/Azure/azure-webjobs-sdk/wiki/EventHub-support

You're also welcome to file feature requests at https://github.com/Azure/azure-webjobs-sdk/issues/

Engineering

In this case I do not understand the sense of an event hub triggered function app. Since myEventHubMessage contains exactly one event I expected the 'new' one coming in (that one that triggered the function). Instead it contains the 'first' one in the queue. – Antisepticize

EventHubTrigger for Function App works as expected (see Ling's answer with my comment). – Antisepticize

@Antisepticize - event hubs uses a cursor. It doesn't read "latest", it reads from the reader's cursor. Put another way; if your writer sends 100 events (and no reader is up yet), and then you launch a reader, it will pick up those 100 events even though they were sent before the reader was launched. – Engineering