CODEWITHSUNDEEP
azure-functionsazure-iot-hubazure-eventhubazure-iot-edge
Ramin
Ramin

Reputation: 390

Offset in Azure EventHubTrigger Function App

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:

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?

Upvotes: 5

Views: 4871

Answers (3)

vullnetyy
vullnetyy

Reputation: 1701

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();
        }
    }
}

Upvotes: 8

Ling Toh
Ling Toh

Reputation: 2474

@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.

Upvotes: 2

Mike S
Mike S

Reputation: 3169

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/

Upvotes: 1

Related Questions