Change Feed Processor: Adds detailed delegate context , stream, and manual checkpointing support

[ealsur]

After 4 months on preview (#2331) this PR moves to GA the new delegates and stream support for Change Feed Processor.

New delegates

We previously had a single delegate to process changes:

public delegate Task ChangesHandler<T>(
	IReadOnlyCollection<T> changes,
	CancellationToken cancellationToken);

In order to address the requests in the linked incidents, we are introducing 4 new delegates:

// Adds the context, similar to the existing handler
public delegate Task ChangeFeedHandler<T>(
	ChangeFeedProcessorContext context,
	IReadOnlyCollection<T> changes,
	CancellationToken cancellationToken);

// Adds support for manual checkpoint on the typed API
public delegate Task ChangeFeedHandlerWithManualCheckpoint<T>(
	ChangeFeedProcessorContext context,
	IReadOnlyCollection<T> changes,
	Func<Task> checkpointAsync,
	CancellationToken cancellationToken);

// Adds support for stream handling
public delegate Task ChangeFeedStreamHandler(
	ChangeFeedProcessorContext context,
	Stream changes,
	CancellationToken cancellationToken);

// Adds supports for manual checkpointing with stream handling
public delegate Task ChangeFeedStreamHandlerWithManualCheckpoint(
	ChangeFeedProcessorContext context,
	Stream changes,
	Func<Task> checkpointAsync,
	CancellationToken cancellationToken);

Container API changes

public abstract ChangeFeedProcessorBuilder GetChangeFeedProcessorBuilder<T>(
	string processorName,
	ChangeFeedHandler<T> onChangesDelegate);

public abstract ChangeFeedProcessorBuilder GetChangeFeedProcessorBuilderWithManualCheckpoint<T>(
	string processorName,
	ChangeFeedHandlerWithManualCheckpoint<T> onChangesDelegate);

public abstract ChangeFeedProcessorBuilder GetChangeFeedProcessorBuilder(
	string processorName,
	ChangeFeedStreamHandler onChangesDelegate);

public abstract ChangeFeedProcessorBuilder GetChangeFeedProcessorBuilderWithManualCheckpoint(
	string processorName,
	ChangeFeedStreamHandlerWithManualCheckpoint onChangesDelegate);

Stream support

Stream APIs allow customers to send the data to other components without serializing it on the Change Feed Processor, this is possible through 2 of the new delegates, the most common one that will handle checkpointing for the user:

public delegate Task ChangeFeedStreamHandler(
	ChangeFeedProcessorContext context,
	Stream changes,
	CancellationToken cancellationToken);

And one where the use can manage checkpoint themselves:

public delegate Task ChangeFeedStreamHandlerWithManualCheckpoint(
	ChangeFeedProcessorContext context,
	Stream changes,
	Func<Task> checkpointAsync,
	CancellationToken cancellationToken);

How to leverage manual checkpoint?

There isn't any extra configuration required on the ChangeFeedProcessorBuilder other than using the delegate that exposes the checkpoint logic. Choosing this delegate is an implicit opt-in on the logic.

This enables users to decide when to checkpoint the progress of a lease. Some users decide to do buffering of changes and once the buffer commits (processes all the changes), then they decide to checkpoint.

Any of the 2 delegates that expose the checkpoint mechanism allows for this, via the call to checkpointAsync, like so:

ChangeFeedProcessor processor = container
	.GetChangeFeedProcessorBuilderWithManualCheckpoint("myProcessorName", 
		async (ChangeFeedProcessorContext context, IReadOnlyCollection<T> changes, Func<Task> checkpointAsync, CancellationToken token) =>
	{
		// manage and process the changes

		// on some condition, trigger the checkpoint
               if (someConditionIsMet)
               {
		    await checkpointAsync();
               }
	})
	.WithInstanceName("<instance name>")
	.WithLeaseContainer(leaseContainer).Build();

What is there in the context?

    public abstract class ChangeFeedProcessorContext
    {
        /// <summary>
        /// Gets the token representative of the current lease from which the changes come from.
        /// </summary>
        public abstract string LeaseToken { get; }

        /// <summary>
        /// Gets the diagnostics related to the service response.
        /// </summary>
        public abstract CosmosDiagnostics Diagnostics { get; }

        /// <summary>
        /// Gets the headers related to the service response that provided the changes.
        /// </summary>
        public abstract Headers Headers { get; }
    }

The context includes information related to changes delivered on the delegate, it includes the identifier of which lease they belong to (this is useful paired with the Container.ChangeFeedEstimator for monitoring). the Headers (that expose RU consumption and SessionToken for extending the session to other components), and the Diagnostics.

Type of change

• New feature (non-breaking change which adds functionality)

cc @bartelink

[j82w]

In v4 SDK changefeed should be put under it's own object like Scripts to keep all the change feed APIs together while reducing the container object bloating.

[kirankumarkolli]

Unsure, scripts namespace is meant for server side scripting (SPROC, Trigger, CRUD etc...).
Sure lets track it for discussion.

[j82w]

I'm meant have a container.ChangeFeed.GetProcessorBuilderWithManualCheckpoint. Then have all the methods and delegates off of the container.ChangeFeed object. That way all the changefeed methods are grouped together.

[ealsur]

Yeah, grouping could allow them to live in a ChangeFeed namespace maybe