azure event hubs "latestEventPosition" not working as intended.

[verikono]

• @azure/event-hubs:
• 5.3.1:
• Ubuntu 20.04:
• nodejs
  • 14.13.1
• browser
  • N/A
• typescript
  • 4.0.3
• Is the bug related to documentation in
  • README.md
  • source code documentation
  • [* ] SDK API docs on https://docs.microsoft.com

Describe the bug
I wrote a mocha test to simple subscribe and publish on a single partition to an eventhub. consumer.subscribe documentation indicates we provide for the first argument an object with 2 functions keyed as "processEvents" and "processError". The second argument is an options object which can include startPosition. This allows for a further object, but can be simplified by the "latestEventPosition" and "earliestEventPosition" imports from '@azure/event-hubs'. when using earliestEventPosition and publishing i get what is expected - all events. when I use latestEventPosition i never get an event published after the method call. Reading the code i found i could provide startPosition with an options object and when providing {enqueuedOn:new Date()} events published after the method call trigger processEvents.

To Reproduce
Steps to reproduce the behavior:

1. See bug description.

Expected behavior
I expect that when I provide "latestEventPosition"to the startPosition option parameter that my subscription will only process events made after the method call.

Screenshots
N/A

Additional context
This may be a server time sync issue and not a code issue. Regardless I'd have thought that event time is measured globally - probably with a epoch timestamp created from GMT/UTC.

[chradek]

@verikono
Thanks for logging this issue. Can you share the code for your test?

Are you calling eventHubProducerClient.sendBatch() right after calling eventHubConsumerClient.subscribe()?

Under the hood, subscribe is non-blocking while it starts a loop to receive messages, so if you call sendBatch directly after calling subscribe, you'll have sent the events before subscribe has had a chance to begin receiving messages. This would be why when you set the startPosition using the enqueuedTime, you get the expected results.

For the purposes of writing a test, there are a couple things you could do.

Wait for processEvents to be invoked before sending the event

You can wait for processEvents to be invoked with an empty list of events before sending from your producer. This would ensure that the receiver is actively receiving events. To speed up your test, you can also change the maxWaitTimeInSeconds to a lower value. This won't be affected by clockSkew like enqueuedTime potentially could.

We have a test that essentially does this here.

Set starting position to the last sequence number in each partition

This is another option that's more resilient than using enqueuedTime since it takes clockSkew out of the equation.
We have an example of how to get the starting position across all partitions in our tests here as well.

[verikono]

Thanks for you hasty response @chradek!

I tried a 10 second delay between eventHubConsumerClient.subscribe() and eventHubProducerClient.sendBatch() (using setTimeout) to allow any transactions to finalise.

I'll put my test battery in a repo at some point today.

Thank you so much for your assistance :)

[verikono]

I apologise for the delayed response to this. Work pulling me left and right at this time.

I've put up a repo for this issue here:
https://github.com/verikono/eventhubs-test

It's a simple mocha test that you can run which shows either my misunderstanding of something or the issue i'm trying to highlight here.

You'll need to create a .env file based off the .env-dev file in the repo root. I left the .vscode configuration in the repo in case it's of use to you.

lines 137 and 138 are of interest. The test will pass if 138 is used but fail if 137 is used. Line 121 is also of interest as typescript believes consumer.subscribes() argument signature is wrong.

Thank you so much for your help here 👍

[chradek]

@verikono
I'm not seeing a delay between when you call subscribe and when you call sendBatch (createBatch will cause a slight delay but not a very long one.)

So, I believe you're still seeing the issue where you send an event before subscribe has had a chance to begin reading events. You could try putting a timed delay in, or wait for processEvents to be invoked once before moving on to your next it() in your tests.

To do the latter, you could do something like this where you pass done into your it() and then invoke it when you're ready to move on:

it(`Instantiate the consumer subscription for latestEventPosition`, (done) => {
  let isDoneCalled = false;
  subscription = consumer.subscribe(
    {
      processEvents: async ( events, context ) => {
        if(!events.length && !isDoneCalled) {
          isDoneCalled = true;
          // Now the next `it()` can be ran.
          done();
          return;
        }
        if(events.length > 1) throw new Error(`Expected exactly 1 event, but got ${events.length}`);
          const event = events[0]
          context.updateCheckpoint(event);
          received_events.push(event);
      },
      processError: err => {
        errors.push(err);
      }
    },
    { 
      startPosition: latestEventPosition,
      maxWaitTimeInSeconds: 1 // Lower the max wait time so we don't wait a minute before ending this test.
    }
  );
});

With regards to the incorrect signature on line 121, this is because all of the process* functions are meant to be async (return a Promise.) They'll technically still work if you don't make them async since you can await a function that doesn't return a promise, but since your processEvents function doesn't return a promise that's why TypeScript is complaining.

[ghost]

Hi, we're sending this friendly reminder because we haven't heard back from you in a while. We need more information about this issue to help address it. Please be sure to give us your input within the next 7 days. If we don't hear back from you within 14 days of this comment the issue will be automatically closed. Thank you!