Skip to content

[Merge]PublicAPIAndMigration #32187

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

Closed
xinlian12 wants to merge 6 commits into from
Closed

[Merge]PublicAPIAndMigration #32187

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
ccfe075
wire up merge public API and add migration
Nov 16, 2022
7d96023
fix failed build
Nov 16, 2022
8bff83f
fix tests
Nov 16, 2022
ee9156c
add more tests
Nov 16, 2022
a37668a
Merge branch 'main' into mergePublicAPIChangeAndMigration
Nov 16, 2022
b940194
Merge branch 'main' into mergePublicAPIChangeAndMigration
Nov 17, 2022
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
27 changes: 27 additions & 0 deletions sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/ChangeFeedProcessorBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -133,6 +133,33 @@ public ChangeFeedProcessorBuilder handleChanges(Consumer<List<JsonNode>> consume
return this;
}

/**
* Sets a consumer function which will be called to process changes for LatestVersion change feed mode.
*
* <!-- src_embed com.azure.cosmos.latestVersionChanges.handleChanges -->
* <pre>
* .handleLatestVersionChanges&#40;docs -&gt; &#123;
* for &#40;JsonNode item : docs&#41; &#123;
* &#47;&#47; Implementation for handling and processing of each JsonNode item goes here
* &#125;
* &#125;&#41;
* </pre>
* <!-- end com.azure.cosmos.latestVersionChanges.handleChanges -->
*
* @param consumer the {@link Consumer} to call for handling the feeds.
* @return current Builder.
*/
@Beta(value = Beta.SinceVersion.V4_40_0, warningText = Beta.PREVIEW_SUBJECT_TO_CHANGE_WARNING)
public ChangeFeedProcessorBuilder handleLatestVersionChanges(Consumer<List<JsonNode>> consumer) {

@xinlian12 xinlian12 Nov 16, 2022

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

should we return JsonNode or ChangeFeedProcessorItem

@FabianMeiswinkel FabianMeiswinkel Nov 17, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

ChangeFeedProcessorItem (to normalizer eventually on public surface area) - but to be honest I don't particularly like the name of that class (probably too late) - ChangeFeedNotification / event/ record would probably describe better what it is. Has the current beta-api naming been discussed with central team already? Or will we have the naming discussion anyway with them? If the former definitely ignore my feedback (not worth changing now). Even if we have not discussed it yet I would only consider it if there are others with similar feedback.

@FabianMeiswinkel FabianMeiswinkel Nov 17, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

I think it would be good to add a comment that this method is a replacement for the old handleChanges (that will eventually be deprecated) - but that the semantics are compatible

checkNotNull(consumer, "Argument 'consumer' can not be null");
checkArgument(this.incrementalModeLeaseConsumer == null, "consumer has already been defined");

this.incrementalModeLeaseConsumer = consumer;
this.changeFeedMode = ChangeFeedMode.INCREMENTAL;
this.leaseVersion = LeaseVersion.EPK_RANGE_BASED_LEASE;
return this;
}

/**
* Sets a consumer function which will be called to process changes for AllVersionsAndDeletes change feed mode.
*
Expand Down
15 changes: 15 additions & 0 deletions sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/HttpConstants.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -275,13 +275,28 @@ public static class HttpHeaders {
// Client Encryption Headers
public static final String IS_CLIENT_ENCRYPTED_HEADER = "x-ms-cosmos-is-client-encrypted";
public static final String INTENDED_COLLECTION_RID_HEADER = "x-ms-cosmos-intended-collection-rid";

// SDK supported capacities headers
public static final String SDK_SUPPORTED_CAPABILITIES = "x-ms-cosmos-sdk-supported-capabilities";
}

public static class A_IMHeaderValues {
public static final String INCREMENTAL_FEED = "Incremental Feed";
public static final String FULL_FIDELITY_FEED = "Full-Fidelity Feed";
}

public static class SDKSupportedCapabilities {
private static final long None = 0; // 0
private static final long PartitionMerge = 1; // 1 << 0

public static final long SUPPORTED_CAPABILITIES;
public static final long SUPPORTED_CAPABILITIES_EXCLUDE_MERGE;
static {
SUPPORTED_CAPABILITIES = PartitionMerge;
SUPPORTED_CAPABILITIES_EXCLUDE_MERGE = None;
}
}

public static class Versions {
public static final String CURRENT_VERSION = "2020-07-15";
public static final String QUERY_VERSION = "1.0";
Expand Down
6 changes: 6 additions & 0 deletions ...smos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/RxDocumentClientImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1648,6 +1648,12 @@ private Mono<RxDocumentServiceRequest> populateHeadersAsync(RxDocumentServiceReq
request.getHeaders().put(HttpConstants.HttpHeaders.API_TYPE, this.apiType.toString());
}

if (!request.getHeaders().containsKey(HttpConstants.HttpHeaders.SDK_SUPPORTED_CAPABILITIES)) {
request.getHeaders().put(
HttpConstants.HttpHeaders.SDK_SUPPORTED_CAPABILITIES,
String.valueOf(HttpConstants.SDKSupportedCapabilities.SUPPORTED_CAPABILITIES));
}

if ((RequestVerb.POST.equals(httpMethod) || RequestVerb.PUT.equals(httpMethod))
&& !request.getHeaders().containsKey(HttpConstants.HttpHeaders.CONTENT_TYPE)) {
request.getHeaders().put(HttpConstants.HttpHeaders.CONTENT_TYPE, RuntimeConstants.MediaTypes.JSON);
Expand Down
2 changes: 2 additions & 0 deletions ...osmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/RxGatewayStoreModel.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,6 +78,8 @@ public RxGatewayStoreModel(
"no-cache");
this.defaultHeaders.put(HttpConstants.HttpHeaders.VERSION,
HttpConstants.Versions.CURRENT_VERSION);
this.defaultHeaders.put(HttpConstants.HttpHeaders.SDK_SUPPORTED_CAPABILITIES,
String.valueOf(HttpConstants.SDKSupportedCapabilities.SUPPORTED_CAPABILITIES));

if (apiType != null){
this.defaultHeaders.put(HttpConstants.HttpHeaders.API_TYPE, apiType.toString());
Expand Down
9 changes: 9 additions & 0 deletions ...mos/src/main/java/com/azure/cosmos/implementation/changefeed/ChangeFeedContextClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,12 +6,14 @@
import com.azure.cosmos.CosmosAsyncDatabase;
import com.azure.cosmos.implementation.PartitionKeyRange;
import com.azure.cosmos.implementation.routing.Range;
import com.azure.cosmos.models.CosmosBulkOperationResponse;
import com.azure.cosmos.models.CosmosChangeFeedRequestOptions;
import com.azure.cosmos.models.CosmosContainerProperties;
import com.azure.cosmos.models.CosmosContainerRequestOptions;
import com.azure.cosmos.models.CosmosContainerResponse;
import com.azure.cosmos.models.CosmosDatabaseRequestOptions;
import com.azure.cosmos.models.CosmosDatabaseResponse;
import com.azure.cosmos.models.CosmosItemOperation;
import com.azure.cosmos.models.CosmosItemRequestOptions;
import com.azure.cosmos.models.CosmosItemResponse;
import com.azure.cosmos.models.CosmosQueryRequestOptions;
Expand Down Expand Up @@ -172,4 +174,11 @@ <T> Mono<CosmosItemResponse<T>> readItem(String itemId, PartitionKey partitionKe
* @return The list of partition key ranges.
*/
Mono<List<PartitionKeyRange>> getOverlappingRanges(Range<String> range);

/**
* Executes flux of operations in Bulk.
* @param operations Flux of operation which will be executed by this container.
* @return A Flux of {@link CosmosBulkOperationResponse} which contains operation and it's response or exception.
*/
Flux<CosmosBulkOperationResponse<Object>> executeBulkOperations(Flux<CosmosItemOperation> operations);
}
9 changes: 9 additions & 0 deletions ...re-cosmos/src/main/java/com/azure/cosmos/implementation/changefeed/LeaseStoreManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.util.List;

/**
* Defines an interface for operations with {@link Lease}.
Expand Down Expand Up @@ -68,6 +69,14 @@ interface LeaseStoreManagerBuilderDefinition {
*/
Mono<Void> delete(Lease lease);

/**
* DELETE all the leases passed in.
*
* @param leases the leases to be removed.
* @return a representation of the deferred computation of this call.
*/
Mono<Void> delete(List<Lease> leases);

/**
* Acquire ownership of the lease.
*
Expand Down
8 changes: 8 additions & 0 deletions ...n/java/com/azure/cosmos/implementation/changefeed/common/ChangeFeedContextClientImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,12 +11,14 @@
import com.azure.cosmos.implementation.PartitionKeyRange;
import com.azure.cosmos.implementation.changefeed.ChangeFeedContextClient;
import com.azure.cosmos.implementation.routing.Range;
import com.azure.cosmos.models.CosmosBulkOperationResponse;
import com.azure.cosmos.models.CosmosChangeFeedRequestOptions;
import com.azure.cosmos.models.CosmosContainerProperties;
import com.azure.cosmos.models.CosmosContainerRequestOptions;
import com.azure.cosmos.models.CosmosContainerResponse;
import com.azure.cosmos.models.CosmosDatabaseRequestOptions;
import com.azure.cosmos.models.CosmosDatabaseResponse;
import com.azure.cosmos.models.CosmosItemOperation;
import com.azure.cosmos.models.CosmosItemRequestOptions;
import com.azure.cosmos.models.CosmosItemResponse;
import com.azure.cosmos.models.CosmosQueryRequestOptions;
Expand Down Expand Up @@ -108,6 +110,12 @@ public Mono<List<PartitionKeyRange>> getOverlappingRanges(Range<String> range) {
.publishOn(this.scheduler);
}

@Override
public Flux<CosmosBulkOperationResponse<Object>> executeBulkOperations(Flux<CosmosItemOperation> operations) {
return this.cosmosContainer.executeBulkOperations(operations)
.publishOn(this.scheduler);
}

@Override
public Flux<FeedResponse<PartitionKeyRange>> readPartitionKeyRangeFeed(String partitionKeyRangesOrCollectionLink, CosmosQueryRequestOptions cosmosQueryRequestOptions) {
return this.documentClient.readPartitionKeyRanges(partitionKeyRangesOrCollectionLink, cosmosQueryRequestOptions)
Expand Down
28 changes: 26 additions & 2 deletions ...src/main/java/com/azure/cosmos/implementation/changefeed/epkversion/BootstrapperImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
import com.azure.cosmos.implementation.CosmosSchedulers;
import com.azure.cosmos.implementation.changefeed.Bootstrapper;
import com.azure.cosmos.implementation.changefeed.LeaseStore;
import com.azure.cosmos.implementation.changefeed.LeaseStoreManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Mono;
Expand All @@ -21,20 +22,28 @@ class BootstrapperImpl implements Bootstrapper {
private final Logger logger = LoggerFactory.getLogger(BootstrapperImpl.class);
private final PartitionSynchronizer synchronizer;
private final LeaseStore leaseStore;
private final LeaseStoreManager pkVersionLeaseStoreManager;

@FabianMeiswinkel FabianMeiswinkel Nov 17, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

I know the term "pkVersion" comes from a previous PR - so non-blocking. I got confused by this prefix - pk - assuming stands for partition key - what is meant here is PKRangeId (PartitionKeyRangeId) - i was wondering why you want a logical partition based lease store - until I realized pk stands for partitioneyrangeid not partitionkey. First thought it is a typo missing the e of epk :-) - anyway like mentioned non-blocking but IMO not very intuitive.

@xinlian12 xinlian12 Nov 17, 2022

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

make sense, maybe pkRangeIdVersion?

@FabianMeiswinkel FabianMeiswinkel Nov 17, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

much better IMO

private final Duration lockTime;
private final Duration sleepTime;

private volatile boolean isInitialized;
private volatile boolean isLockAcquired;

public BootstrapperImpl(PartitionSynchronizer synchronizer, LeaseStore leaseStore, Duration lockTime, Duration sleepTime) {
public BootstrapperImpl(
PartitionSynchronizer synchronizer,
LeaseStore leaseStore,
Duration lockTime,
Duration sleepTime,
LeaseStoreManager pkVersionLeaseStoreManager) {

checkNotNull(synchronizer, "Argument 'synchronizer' can not be null");
checkNotNull(leaseStore, "Argument 'leaseStore' can not be null");
checkArgument(lockTime != null && this.isPositive(lockTime), "lockTime should be non-null and positive");
checkArgument(sleepTime != null && this.isPositive(sleepTime), "sleepTime should be non-null and positive");

this.synchronizer = synchronizer;
this.leaseStore = leaseStore;
this.pkVersionLeaseStoreManager = pkVersionLeaseStoreManager;
this.lockTime = lockTime;
this.sleepTime = sleepTime;

Expand Down Expand Up @@ -66,7 +75,18 @@ public Mono<Void> initialize() {
logger.info("Another instance is initializing the store");
return Mono.just(isLockAcquired).delayElement(this.sleepTime, CosmosSchedulers.COSMOS_PARALLEL);
} else {
return this.synchronizer.createMissingLeases()
return this.shouldBootstrapFromPkVersionLeases()
.flatMap(shouldBootstrapFromPkVersion -> {
if (shouldBootstrapFromPkVersion) {
return this.pkVersionLeaseStoreManager.getAllLeases()
.collectList()
.flatMap(pkVersionLeases -> {
return this.synchronizer.createMissingLeases(pkVersionLeases)
.then(this.pkVersionLeaseStoreManager.delete(pkVersionLeases)); // after bootstrapping, we are going to delete all th pk version leases
});
}
return this.synchronizer.createMissingLeases();
})
.then(this.leaseStore.markInitialized());
}
})
Expand All @@ -85,4 +105,8 @@ public Mono<Void> initialize() {
.repeat(() -> !this.isInitialized)
.then();
}

private Mono<Boolean> shouldBootstrapFromPkVersionLeases() {
return this.pkVersionLeaseStoreManager == null ? Mono.just(false) : this.pkVersionLeaseStoreManager.isInitialized();
}
}
52 changes: 48 additions & 4 deletions ...va/com/azure/cosmos/implementation/changefeed/epkversion/ChangeFeedProcessorImplBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@

package com.azure.cosmos.implementation.changefeed.epkversion;

import com.azure.cosmos.BridgeInternal;
import com.azure.cosmos.ChangeFeedProcessor;
import com.azure.cosmos.ConsistencyLevel;
import com.azure.cosmos.CosmosAsyncContainer;
Expand Down Expand Up @@ -67,9 +68,12 @@ public abstract class ChangeFeedProcessorImplBase<T> implements ChangeFeedProces
private final Scheduler scheduler;

private volatile String databaseResourceId;
private volatile String databaseId;
private volatile String collectionResourceId;
private volatile String collectionId;
private PartitionLoadBalancingStrategy loadBalancingStrategy;
private LeaseStoreManager leaseStoreManager;
private LeaseStoreManager pkVersionLeaseStoreManager;
private HealthMonitor healthMonitor;
private volatile PartitionManager partitionManager;

Expand Down Expand Up @@ -257,12 +261,14 @@ private Mono<ChangeFeedProcessor> initializeCollectionPropertiesForBuild() {
.readDatabase(this.feedContextClient.getDatabaseClient(), null)
.map(databaseResourceResponse -> {
this.databaseResourceId = databaseResourceResponse.getProperties().getResourceId();
this.databaseId = databaseResourceResponse.getProperties().getId();
return this.databaseResourceId;
})
.flatMap( id -> this.feedContextClient
.readContainer(this.feedContextClient.getContainerClient(), null)
.map(documentCollectionResourceResponse -> {
this.collectionResourceId = documentCollectionResourceResponse.getProperties().getResourceId();
this.collectionId = documentCollectionResourceResponse.getProperties().getId();
return this;
}));
}
Expand All @@ -287,6 +293,19 @@ private Mono<LeaseStoreManager> getLeaseStoreManager() {
.requestOptionsFactory(requestOptionsFactory)
.hostName(this.hostName)
.build();

if (this.canBootstrapFromPkVersionLeaseStore()) {
String pkVersionLeasePrefix = this.getPkVersionLeasePrefix();
this.pkVersionLeaseStoreManager =
com.azure.cosmos.implementation.changefeed.pkversion.LeaseStoreManagerImpl.builder()
.leasePrefix(pkVersionLeasePrefix)
.leaseCollectionLink(this.leaseContextClient.getContainerClient())
.leaseContextClient(this.leaseContextClient)
.requestOptionsFactory(requestOptionsFactory)
.hostName(this.hostName)
.build();
}

return Mono.just(this.leaseStoreManager);
});
}
Expand Down Expand Up @@ -319,20 +338,45 @@ private String getLeasePrefix() {
this.collectionResourceId);
}

private String getPkVersionLeasePrefix() {
String optionsPrefix = this.changeFeedProcessorOptions.getLeasePrefix();

if (optionsPrefix == null) {
optionsPrefix = "";
}

URI uri = this.feedContextClient.getServiceEndpoint();

return String.format(
"%s%s_%s_%s",
optionsPrefix,
uri.getHost(),
this.databaseId,
this.collectionId);
}

abstract Class<T> getPartitionProcessorItemType();
abstract boolean canBootstrapFromPkVersionLeaseStore();

private Mono<PartitionManager> buildPartitionManager(LeaseStoreManager leaseStoreManager) {
CheckpointerObserverFactory<T> factory = new CheckpointerObserverFactory<>(this.observerFactory, new CheckpointFrequency());

PartitionSynchronizerImpl synchronizer = new PartitionSynchronizerImpl(
this.feedContextClient,
this.feedContextClient.getContainerClient(),
BridgeInternal.extractContainerSelfLink(this.feedContextClient.getContainerClient()),
leaseStoreManager,
leaseStoreManager,
DEFAULT_DEGREE_OF_PARALLELISM,
DEFAULT_QUERY_PARTITIONS_MAX_BATCH_SIZE);

Bootstrapper bootstrapper = new BootstrapperImpl(synchronizer, leaseStoreManager, this.lockTime, this.sleepTime);
DEFAULT_QUERY_PARTITIONS_MAX_BATCH_SIZE,
this.changeFeedProcessorOptions,
this.changeFeedMode);

Bootstrapper bootstrapper = new BootstrapperImpl(
synchronizer,
leaseStoreManager,
this.lockTime,
this.sleepTime,
this.pkVersionLeaseStoreManager);
PartitionSupervisorFactory partitionSupervisorFactory = new PartitionSupervisorFactoryImpl<>(
factory,
leaseStoreManager,
Expand Down
6 changes: 6 additions & 0 deletions ...zure/cosmos/implementation/changefeed/epkversion/FullFidelityChangeFeedProcessorImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,4 +37,10 @@ CosmosChangeFeedRequestOptions createRequestOptionsForProcessingFromNow(FeedRang
Class<ChangeFeedProcessorItem> getPartitionProcessorItemType() {
return ChangeFeedProcessorItem.class;
}

@Override
boolean canBootstrapFromPkVersionLeaseStore() {
// pkVersion does not support fullFidelity mode
return false;
}
}
5 changes: 5 additions & 0 deletions ...azure/cosmos/implementation/changefeed/epkversion/IncrementalChangeFeedProcessorImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,4 +35,9 @@ CosmosChangeFeedRequestOptions createRequestOptionsForProcessingFromNow(FeedRang
Class<JsonNode> getPartitionProcessorItemType() {
return JsonNode.class;
}

@Override
boolean canBootstrapFromPkVersionLeaseStore() {
return true;
}
}
6 changes: 6 additions & 0 deletions ...ain/java/com/azure/cosmos/implementation/changefeed/epkversion/LeaseStoreManagerImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@

import java.time.Duration;
import java.util.Collections;
import java.util.List;

import static com.azure.cosmos.implementation.guava25.base.Preconditions.checkArgument;
import static com.azure.cosmos.implementation.guava25.base.Preconditions.checkNotNull;
Expand Down Expand Up @@ -205,6 +206,11 @@ public Mono<Void> delete(Lease lease) {
.then();
}

@Override
public Mono<Void> delete(List<Lease> leases) {
throw new UnsupportedOperationException("Delete with leases are not supported for Change Feed epk version wire format");

@FabianMeiswinkel FabianMeiswinkel Nov 17, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

On first glance I would ask why you even added this to the base class? Handling UnsupportedException to recognize seems worse than letting compiler to pick it up? But maybe further review will help to answer that question...

}

@Override
public Mono<Lease> acquire(Lease lease) {
if (lease == null) {
Expand Down
Loading