Board Review: azure-eventhub receive_batch

[YijunXieMS]

The Basics

• Service team responsible for the client library: Event Hubs
• Link to documentation describing the service: https://azure.microsoft.com/en-us/services/event-hubs/
• Contact email (if service team, provide PM and Dev Lead): SeongJoon Kwak, Javier Fernandez & Shubha Vijayasarathy

About this client library

• Name of the client library: azure-eventhub
• Languages for this review: Python, Java
• Link to the service REST APIs: N/A

Artifacts required (per language)

Python

• Upload the api as a Draft PR. Link to PR: https://github.com/Azure/azure-sdk-for-python/pull/9636/
• Link to samples for champion scenarios

Java

• API review
• Sample user experience

Champion Scenarios

A champion scenario is a use case that the consumer of the client library is commonly expected to perform. Champion scenarios are used to ensure the developer experience is exemplary for the common cases. You need to show the entire code sample (including error handling, as an example) for the champion scenarios.

• Champion Scenario 1:
  • Describe the champion scenario: Receive a batch of events, process them and save checkpoint using the last event. Users can process events and checkpoint by batch size, by time, or whichever one comes first by setting different values of max_batch_size and max_wait_time.
  • Estimate the percentage of developers using the service who would use the champion scenario: Most developers will use it.
  • Link to the code sample

[adrianhall]

Pre-assigning to @johanste for initial review. He will determine if a full board review is needed.

[johanste]

One architectural cross-cutting question; under what circumstances (if any) do we call the event handler with an empty list of events.

[johanste]

This should be a quick review (hopefully)

[adrianhall]

Notes from Arch Board: https://msit.microsoftstream.com/video/4dfba91e-4092-42a3-acfd-3db666af8eef

Java Review

• Java sync client already has batching - async client uses Flux semantics.
• DECISION Don't do the receiveBatch convenience method.
• Why isn't updateCheckpoint() automatically called? It's because the app may not have finished processing the events. (async)
• Concerns around heartbeat - JS would like heartbeat to be opt-in. James said T1 had heartbeat, but off by default. No information on how often it was used.
• DECISION Heartbeat is a good feature, should be opt-in. Feature team should design appropriately.
• SIngle event processor also has same suggested semantics. Should we have heartbeat?
• DECISION We should have the same heartbeat features on the single event receiver as on the batch event receiver.

Python Review:

• Potential new name for receive_batch (receive_in_batches perhaps) - not a better name.
• Would like new name for enable_callback_when_no_event Can't we use max_wait_time of None?
• Is max_wait_time appropriate name? Maybe max_processing_delay?
• It's two different timings? (latency vs. heartbeat time)
• RECOMMENDATION Time between heartbeats & max time to process after event arrived should probably be separate timings. Make an informed design decision on this.

[JamesBirdsall]

A very important reason why the client should never checkpoint automatically: some customers don’t use checkpoints. At all. When they open receivers, they start at LATEST, or some fixed time before NOW, and go from there. It took us a while to learn that lesson, and I want to be sure it doesn’t get lost in the transition to track 2.

The semantics of the heartbeat were simple in track 1 because track 1 did not have batching. EventProcessorHost effectively has a receive loop in it, and the track 1 receive call returns as soon as at least one message is available. EPH takes whatever the receive call returned and passes it to the user’s callback, then goes back to the top of the loop. If the receive call times out, it returns null, and the loop either translates that into an empty message list for the callback (if heartbeat is on) or skips the callback (if heartbeat is off). So with a receive timeout of one minute, if heartbeat is on, the result is straightforward: the maximum time between one call of the callback returning and the next starting is one minute. Reproducing that semantic for the single-event callback should be simple because the semantics of receiving one event at a time are about the same as the track 1 receive. Reproducing it for the batch callback is going to be complicated.

[annatisch]

Final design for Python:

receive_batch(
    on_event_batch,  # type: Callable[PartitionContext, List[EventData]]
    max_batch_size=300,  # type: int
    max_wait_time=None,  # type: Optional[int]
    **kwargs
)

Semantics are as follows:

• The default behaviour: max_wait_time is set to None, in which case we will wait indefinitely for events to arrive without calling the callback. When one or more events arrive we will call the callback immediately without waiting to reach to maximum batch size. This behaviour will also apply if max_wait_time is set to 0. The priority here is performance, and low message latency.
• If a max_wait_time value is set, the processor will wait at most the specified number of seconds to either fill the message batch to the maximum size, or call the callback regardless of how many messages we have. If we have received no messages in this interval, the callback will be called with an empty list. This setting provides users with both a regular heartbeat/checkpointing mechanism as well as the ability to prioritize full batches over low message latency.
• The max_wait_time parameter will also be added to the single event receive, with the same behaviour and same default.

Unsupported scenarios:

• The only scenario not supported is combining both the heartbeat behaviour with the low message latency behaviour (return new events immediately). We are satisfied that should this scenario become a genuine customer ask, we can support it later with non-breaking changes.

[srnagar]

Final API for Java:

EventProcessorClientBuilder.java will have the following APIs added:

// Single event receive with hearbeat
public EventProcessorClientBuilder processEvent(Consumer<EventContext> processEvent, 
        Duration maxWaitTime) {}

// Batch receive
public EventProcessorClientBuilder processEventBatch(Consumer<List<EventContext>> processEventBatch, 
        int maxBatchSize) {}
public EventProcessorClientBuilder processEventBatch(Consumer<List<EventContext>> processEventBatch, 
        int maxBatchSize, Duration maxWaitTime) {}

The semantics will be the same as Python.

API View - https://apiview.dev/Assemblies/Review/7d5c89c868424d45bbd5319d5f5e96b8?diffRevisionId=0d96e4d2e8bc436a8ef3f0aacc49672c

cc: @JonathanGiles

[bterlson]

These proposed changes LGTM! :shipit:

[johanste]

+1 - :shipit:

[KrzysztofCwalina]

Looks good!