Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[Preview] Change Feed Processor: Refactors manual checkpoint API #2488

Merged
merged 21 commits into from
Jun 3, 2021
Merged

[Preview] Change Feed Processor: Refactors manual checkpoint API #2488

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
21 commits
Select commit Hold shift + click to select a range
aa7b997
Refactoring contract
ealsur May 20, 2021
11dc624
contract
ealsur May 20, 2021
b5c8292
conditional
ealsur May 20, 2021
7e6a32e
Merge branch 'master' into users/ealsur/checkpointreview
ealsur May 20, 2021
abb4ee2
Missed one
ealsur May 20, 2021
e26b7bd
Merge branch 'users/ealsur/checkpointreview' of https://github.com/Az…
ealsur May 20, 2021
a3170c9
Merge branch 'master' into users/ealsur/checkpointreview
ealsur May 20, 2021
d8eb2d6
yml fix
ealsur May 20, 2021
97d5cb4
Merge branch 'users/ealsur/checkpointreview' of https://github.com/Az…
ealsur May 20, 2021
3f08985
Merge branch 'master' into users/ealsur/checkpointreview
ealsur May 20, 2021
f0ca950
test
ealsur May 21, 2021
f539192
Merge branch 'users/ealsur/checkpointreview' of https://github.com/Az…
ealsur May 21, 2021
021f861
emulator csproj
ealsur May 21, 2021
101c99b
more tests
ealsur May 21, 2021
8659bf0
Merge branch 'master' into users/ealsur/checkpointreview
j82w May 26, 2021
b9b4c84
Merge branch 'master' into users/ealsur/checkpointreview
ealsur May 27, 2021
2f1d638
Merge branch 'master' into users/ealsur/checkpointreview
ealsur May 28, 2021
a83499f
Comments
ealsur May 28, 2021
2b833f2
Merge branch 'master' into users/ealsur/checkpointreview
j82w Jun 2, 2021
29608ee
Merge branch 'master' into users/ealsur/checkpointreview
ealsur Jun 2, 2021
03daf80
Merge branch 'master' into users/ealsur/checkpointreview
ealsur Jun 3, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 1 addition & 5 deletions Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/Observers/AutoCheckpointer.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,11 +35,7 @@ public override async Task ProcessChangesAsync(
{
await this.observer.ProcessChangesAsync(context, stream, cancellationToken).ConfigureAwait(false);

(bool isSuccess, Exception exception) = await context.TryCheckpointAsync();
if (!isSuccess)
{
throw exception;
}
await context.CheckpointAsync();
}
}
}
9 changes: 4 additions & 5 deletions Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/Observers/ChangeFeedObserverContextCore.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,12 +37,11 @@ internal ChangeFeedObserverContextCore(

public Headers Headers => this.responseMessage.Headers;

public async Task<(bool isSuccess, Exception error)> TryCheckpointAsync()
public async Task CheckpointAsync()
{
try
{
await this.checkpointer.CheckpointPartitionAsync(this.responseMessage.Headers.ContinuationToken);
return (isSuccess: true, error: null);
ealsur marked this conversation as resolved.
Show resolved Hide resolved
}
catch (LeaseLostException leaseLostException)
{
Expand All @@ -55,11 +54,11 @@ internal ChangeFeedObserverContextCore(
trace: NoOpTrace.Singleton,
error: null,
innerException: leaseLostException);
return (isSuccess: false, error: cosmosException);
throw cosmosException;
}
catch (Exception exception)
catch
{
return (isSuccess: false, error: exception);
throw;
ealsur marked this conversation as resolved.
Show resolved Hide resolved
}
}

Expand Down
4 changes: 2 additions & 2 deletions Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/Observers/ChangeFeedObserverFactoryCore.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,7 @@ private Task ChangesStreamHandlerAsync(
return this.onChanges(context, stream, cancellationToken);
}

return this.onChangesWithManualCheckpoint(context, stream, context.TryCheckpointAsync, cancellationToken);
return this.onChangesWithManualCheckpoint(context, stream, context.CheckpointAsync, cancellationToken);
}
}

Expand Down Expand Up @@ -108,7 +108,7 @@ private Task ChangesStreamHandlerAsync(
return this.onChanges(context, changes, cancellationToken);
}

return this.onChangesWithManualCheckpoint(context, changes, context.TryCheckpointAsync, cancellationToken);
return this.onChangesWithManualCheckpoint(context, changes, context.CheckpointAsync, cancellationToken);
}

private IReadOnlyCollection<T> AsIReadOnlyCollection(Stream stream)
Expand Down
36 changes: 21 additions & 15 deletions Microsoft.Azure.Cosmos/src/Resource/Container/Container.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1482,22 +1482,25 @@ public delegate Task ChangeFeedHandler<T>(
/// </summary>
/// <param name="context">The context related to the changes.</param>
/// <param name="changes">The changes that happened.</param>
/// <param name="tryCheckpointAsync">A task representing an asynchronous checkpoint on the progress of a lease.</param>
/// <param name="checkpointAsync">A task representing an asynchronous checkpoint on the progress of a lease.</param>
/// <param name="cancellationToken">A cancellation token representing the current cancellation status of the <see cref="ChangeFeedProcessor"/> instance.</param>
/// <returns>A <see cref="Task"/> representing the asynchronous operation that is going to be done with the changes.</returns>
/// <example>
/// <code language="c#">
/// <![CDATA[
/// (ChangeFeedProcessorContext context, IReadOnlyCollection<T> changes, Func<Task<(bool isSuccess, CosmosException error)>> tryCheckpointAsync, CancellationToken cancellationToken) =>
/// (ChangeFeedProcessorContext context, IReadOnlyCollection<T> changes, Func<Task> checkpointAsync, CancellationToken cancellationToken) =>
/// {
/// // consume changes
///
/// // On certain condition, we can checkpoint
/// (bool isSuccess, Exception error) checkpointResult = await tryCheckpointAsync();
/// if (!isSuccess)
/// try
/// {
/// // log error, could not checkpoint
/// throw error;
/// await checkpointAsync();
/// }
/// catch (Exception exception)
ealsur marked this conversation as resolved.
Show resolved Hide resolved
/// {
/// // error occurred
/// throw;
/// }
/// }
/// ]]>
Expand All @@ -1506,7 +1509,7 @@ public delegate Task ChangeFeedHandler<T>(
public delegate Task ChangeFeedHandlerWithManualCheckpoint<T>(
ChangeFeedProcessorContext context,
IReadOnlyCollection<T> changes,
Func<Task<(bool isSuccess, Exception error)>> tryCheckpointAsync,
Func<Task> checkpointAsync,
CancellationToken cancellationToken);

/// <summary>
Expand All @@ -1526,31 +1529,34 @@ public delegate Task ChangeFeedStreamHandler(
/// </summary>
/// <param name="context">The context related to the changes.</param>
/// <param name="changes">The changes that happened.</param>
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should this be something a little more obvious like changesPayload?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Well, in this case, it is actually the list (IReadOnlyCollection<T>) of changes that happened. Those changes in the collection could be inserts or updates, but they are still changes. Not sure how the word payload makes it more clear?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Also, the current API that is GA uses changes as name, so I rather keep the same as we have no feedback that indicates the name brings confusion?

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Then it's fine. It just seemed a little off that a stream would be called changes.

/// <param name="tryCheckpointAsync">A task representing an asynchronous checkpoint on the progress of a lease.</param>
/// <param name="checkpointAsync">A task representing an asynchronous checkpoint on the progress of a lease.</param>
/// <param name="cancellationToken">A cancellation token representing the current cancellation status of the <see cref="ChangeFeedProcessor"/> instance.</param>
/// <returns>A <see cref="Task"/> representing the asynchronous operation that is going to be done with the changes.</returns>
/// <example>
/// <code language="c#">
/// <![CDATA[
/// (ChangeFeedProcessorContext context, Stream stream, Func<Task<(bool isSuccess, CosmosException error)>> tryCheckpointAsync, CancellationToken cancellationToken) =>
/// (ChangeFeedProcessorContext context, Stream stream, Func<Task> checkpointAsync, CancellationToken cancellationToken) =>
/// {
/// // consume stream
///
/// // On certain condition, we can checkpoint
/// (bool isSuccess, Exception error) checkpointResult = await tryCheckpointAsync();
/// if (!isSuccess)
/// try
ealsur marked this conversation as resolved.
Show resolved Hide resolved
/// {
/// // log error, could not checkpoint
/// throw error;
/// await checkpointAsync();
/// }
/// }
/// catch (Exception exception)
/// {
/// // error occurred
/// throw;
/// }
/// }/// }
/// ]]>
/// </code>
/// </example>
public delegate Task ChangeFeedStreamHandlerWithManualCheckpoint(
ChangeFeedProcessorContext context,
Stream changes,
Func<Task<(bool isSuccess, Exception error)>> tryCheckpointAsync,
ealsur marked this conversation as resolved.
Show resolved Hide resolved
Func<Task> checkpointAsync,
CancellationToken cancellationToken);

/// <summary>
Expand Down
4 changes: 2 additions & 2 deletions Microsoft.Azure.Cosmos/src/Resource/Container/ContainerInternal.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -183,7 +183,7 @@ public delegate Task ChangeFeedHandler<T>(
public delegate Task ChangeFeedHandlerWithManualCheckpoint<T>(
ChangeFeedProcessorContext context,
IReadOnlyCollection<T> changes,
Func<Task<(bool isSuccess, Exception error)>> tryCheckpointAsync,
Func<Task> checkpointAsync,
CancellationToken cancellationToken);

public delegate Task ChangeFeedStreamHandler(
Expand All @@ -194,7 +194,7 @@ public delegate Task ChangeFeedStreamHandler(
public delegate Task ChangeFeedStreamHandlerWithManualCheckpoint(
ChangeFeedProcessorContext context,
Stream changes,
Func<Task<(bool isSuccess, Exception error)>> tryCheckpointAsync,
Func<Task> checkpointAsync,
CancellationToken cancellationToken);

public abstract ChangeFeedProcessorBuilder GetChangeFeedProcessorBuilder<T>(
Expand Down
6 changes: 2 additions & 4 deletions ....Azure.Cosmos/tests/Microsoft.Azure.Cosmos.EmulatorTests/ChangeFeed/DynamicStreamTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -93,7 +93,7 @@ public async Task TestWithRunningProcessor_WithManualCheckpoint()
int processedDocCount = 0;
string accumulator = string.Empty;
ChangeFeedProcessor processor = this.Container
.GetChangeFeedProcessorBuilderWithManualCheckpoint("test", async (ChangeFeedProcessorContext context, Stream stream, Func<Task<(bool isSuccess, Exception error)>> tryCheckpointAsync, CancellationToken token) =>
.GetChangeFeedProcessorBuilderWithManualCheckpoint("test", async (ChangeFeedProcessorContext context, Stream stream, Func<Task> checkpointAsync, CancellationToken token) =>
{
this.ValidateContext(context);

Expand All @@ -113,9 +113,7 @@ public async Task TestWithRunningProcessor_WithManualCheckpoint()
if (processedDocCount == 1)
{
// Checkpointing on the first document to be able to have a point to rollback to
(bool isSuccess, Exception exception) = await tryCheckpointAsync();
Assert.IsTrue(isSuccess);
Assert.IsNull(exception);
await checkpointAsync();
}

if (processedDocCount == 12)
Expand Down
6 changes: 2 additions & 4 deletions Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.EmulatorTests/ChangeFeed/DynamicTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,7 +89,7 @@ public async Task TestWithRunningProcessor_WithManualCheckpoint()
int processedDocCount = 0;
string accumulator = string.Empty;
ChangeFeedProcessor processor = this.Container
.GetChangeFeedProcessorBuilderWithManualCheckpoint("test", async (ChangeFeedProcessorContext context, IReadOnlyCollection<dynamic> docs, Func<Task<(bool isSuccess, Exception error)>> tryCheckpointAsync, CancellationToken token) =>
.GetChangeFeedProcessorBuilderWithManualCheckpoint("test", async (ChangeFeedProcessorContext context, IReadOnlyCollection<dynamic> docs, Func<Task> checkpointAsync, CancellationToken token) =>
{
this.ValidateContext(context);
processedDocCount += docs.Count();
Expand All @@ -106,9 +106,7 @@ public async Task TestWithRunningProcessor_WithManualCheckpoint()

if (processedDocCount == 1) {
// Checkpointing on the first document to be able to have a point to rollback to
(bool isSuccess, Exception exception) = await tryCheckpointAsync();
Assert.IsTrue(isSuccess);
Assert.IsNull(exception);
await checkpointAsync();
}

if (processedDocCount == 12)
Expand Down
18 changes: 5 additions & 13 deletions ...osmos/tests/Microsoft.Azure.Cosmos.Tests/ChangeFeed/ChangeFeedObserverContextCoreTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,9 +42,7 @@ public async Task TryCheckpoint_OnSuccess()
checkpointer.Setup(c => c.CheckpointPartitionAsync(It.Is<string>(s => s == continuation))).Returns(Task.CompletedTask);
ChangeFeedObserverContextCore changeFeedObserverContextCore = new ChangeFeedObserverContextCore(leaseToken, responseMessage, checkpointer.Object);

(bool isSuccess, Exception exception) = await changeFeedObserverContextCore.TryCheckpointAsync();
Assert.IsTrue(isSuccess);
Assert.IsNull(exception);
await changeFeedObserverContextCore.CheckpointAsync();
}

[TestMethod]
Expand All @@ -58,10 +56,8 @@ public async Task TryCheckpoint_OnLeaseLost()
checkpointer.Setup(c => c.CheckpointPartitionAsync(It.Is<string>(s => s == continuation))).ThrowsAsync(new LeaseLostException());
ChangeFeedObserverContextCore changeFeedObserverContextCore = new ChangeFeedObserverContextCore(leaseToken, responseMessage, checkpointer.Object);

(bool isSuccess, Exception exception) = await changeFeedObserverContextCore.TryCheckpointAsync();
Assert.IsFalse(isSuccess);
Assert.IsNotNull(exception);
Assert.AreEqual(HttpStatusCode.PreconditionFailed, (exception as CosmosException).StatusCode);
CosmosException exception = await Assert.ThrowsExceptionAsync<CosmosException>(() => changeFeedObserverContextCore.CheckpointAsync());
Assert.AreEqual(HttpStatusCode.PreconditionFailed, exception.StatusCode);
}

[TestMethod]
Expand All @@ -76,9 +72,7 @@ public async Task TryCheckpoint_OnCosmosException()
checkpointer.Setup(c => c.CheckpointPartitionAsync(It.Is<string>(s => s == continuation))).ThrowsAsync(cosmosException);
ChangeFeedObserverContextCore changeFeedObserverContextCore = new ChangeFeedObserverContextCore(leaseToken, responseMessage, checkpointer.Object);

(bool isSuccess, Exception exception) = await changeFeedObserverContextCore.TryCheckpointAsync();
Assert.IsFalse(isSuccess);
Assert.IsNotNull(exception);
CosmosException exception = await Assert.ThrowsExceptionAsync<CosmosException>(() => changeFeedObserverContextCore.CheckpointAsync());
Assert.ReferenceEquals(cosmosException, exception);
}

Expand All @@ -94,9 +88,7 @@ public async Task TryCheckpoint_OnUnknownException()
checkpointer.Setup(c => c.CheckpointPartitionAsync(It.Is<string>(s => s == continuation))).ThrowsAsync(cosmosException);
ChangeFeedObserverContextCore changeFeedObserverContextCore = new ChangeFeedObserverContextCore(leaseToken, responseMessage, checkpointer.Object);

(bool isSuccess, Exception exception) = await changeFeedObserverContextCore.TryCheckpointAsync();
Assert.IsFalse(isSuccess);
Assert.IsNotNull(exception);
NotImplementedException exception = await Assert.ThrowsExceptionAsync<NotImplementedException>(() => changeFeedObserverContextCore.CheckpointAsync());
Assert.ReferenceEquals(cosmosException, exception);
}
}
Expand Down
4 changes: 2 additions & 2 deletions ...osmos/tests/Microsoft.Azure.Cosmos.Tests/ChangeFeed/ChangeFeedObserverFactoryCoreTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,7 +77,7 @@ Task changesHandler(ChangeFeedProcessorContext context, IReadOnlyCollection<dyna
public async Task WhenDelegateIsTyped_Manual()
{
bool executed = false;
Task changesHandler(ChangeFeedProcessorContext context, IReadOnlyCollection<dynamic> docs, Func<Task<(bool isSuccess, Exception exception)>> tryCheckpointAsync, CancellationToken token)
Task changesHandler(ChangeFeedProcessorContext context, IReadOnlyCollection<dynamic> docs, Func<Task> checkpointAsync, CancellationToken token)
{
Assert.AreEqual(1, docs.Count);
Assert.AreEqual("Test", docs.First().id.ToString());
Expand Down Expand Up @@ -128,7 +128,7 @@ public async Task WhenDelegateIsStream_Manual()
{
ResponseMessage responseMessage = this.BuildResponseMessage();
bool executed = false;
Task changesHandler(ChangeFeedProcessorContext context, Stream stream, Func<Task<(bool isSuccess, Exception exception)>> tryCheckpointAsync, CancellationToken token)
Task changesHandler(ChangeFeedProcessorContext context, Stream stream, Func<Task> checkpointAsync, CancellationToken token)
{
Assert.ReferenceEquals(responseMessage.Content, stream);
executed = true;
Expand Down
Loading