in sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/changefeed/pkversion/PartitionProcessorImpl.java [86:280]
public Mono<Void> run(CancellationToken cancellationToken) {
logger.info("Partition {}: processing task started with owner {}.", this.lease.getLeaseToken(), this.lease.getOwner());
this.hasMoreResults = true;
this.checkpointer.setCancellationToken(cancellationToken);
// We only calculate/get the throughput control group config for the feed range at the beginning
// Only split/merge will impact the leases <-> partitionKeyRange mapping
// When split/merge happens, the processor for the current lease will be closed
// and a new processor will be created during load balancing stage
ThroughputControlGroupConfig throughputControlGroupConfigForFeedRange = this.tryGetThroughputControlConfigForFeedRange(this.lease);
return Flux.just(this)
.flatMap( value -> {
if (cancellationToken.isCancellationRequested()) {
return Flux.empty();
}
// If there are still changes need to be processed, fetch right away
// If there are no changes, wait pollDelay time then try again
if(this.hasMoreResults && this.resultException == null) {
return Flux.just(value);
}
Instant stopTimer = Instant.now().plus(this.settings.getFeedPollDelay());
return Mono.just(value)
.delayElement(Duration.ofMillis(100), CosmosSchedulers.COSMOS_PARALLEL)
.repeat( () -> {
Instant currentTime = Instant.now();
return !cancellationToken.isCancellationRequested() && currentTime.isBefore(stopTimer);
}).last();
})
.flatMap(value -> {
if (throughputControlGroupConfigForFeedRange != null) {
this.options.setThroughputControlGroupName(throughputControlGroupConfigForFeedRange.getGroupName());
}
return this.documentClient.createDocumentChangeFeedQuery(
this.settings.getCollectionSelfLink(),
this.options,
JsonNode.class);
})
.flatMap(documentFeedResponse -> {
if (cancellationToken.isCancellationRequested()) return Flux.error(new TaskCancelledException());
final String continuationToken = documentFeedResponse.getContinuationToken();
final ChangeFeedState continuationState = ChangeFeedState.fromString(continuationToken);
checkNotNull(continuationState, "Argument 'continuationState' must not be null.");
checkArgument(
continuationState
.getContinuation()
.getContinuationTokenCount() == 1,
"For ChangeFeedProcessor the continuation state should always have one range/continuation");
this.lastServerContinuationToken = continuationState
.getContinuation()
.getCurrentContinuationToken()
.getToken();
this.hasMoreResults = !ModelBridgeInternal.noChanges(documentFeedResponse);
if (documentFeedResponse.getResults() != null && documentFeedResponse.getResults().size() > 0) {
logger.info("Partition {}: processing {} feeds with owner {}.", this.lease.getLeaseToken(), documentFeedResponse.getResults().size(), this.lease.getOwner());
return this.dispatchChanges(documentFeedResponse, continuationState)
.doOnError(throwable -> logger.debug(
"Exception was thrown from thread {}",
Thread.currentThread().getId(), throwable))
.doOnSuccess((Void) -> {
this.options =
CosmosChangeFeedRequestOptions
.createForProcessingFromContinuation(continuationToken);
if (cancellationToken.isCancellationRequested()) throw new TaskCancelledException();
});
} else {
// still need to checkpoint with the new continuation token
return this.checkpointer.checkpointPartition(continuationState)
.doOnError(throwable -> {
logger.debug(
"Failed to checkpoint partition {} from thread {}",
this.lease.getLeaseToken(),
Thread.currentThread().getId(),
throwable);
})
.flatMap(lease -> {
this.options =
CosmosChangeFeedRequestOptions
.createForProcessingFromContinuation(continuationToken);
if (cancellationToken.isCancellationRequested()) {
return Mono.error(new TaskCancelledException());
}
return Mono.empty();
});
}
})
.doOnComplete(() -> {
if (this.options.getMaxItemCount() != this.settings.getMaxItemCount()) {
this.options.setMaxItemCount(this.settings.getMaxItemCount()); // Reset after successful execution.
}
})
.onErrorResume(throwable -> {
if (throwable instanceof CosmosException) {
// NOTE - the reason why it is safe to access the this.lastServerContinuationToken
// below in a tread-safe manner is because the CosmosException would never be thrown
// form the flatMap-section above (but only from the "source" (the flatMap-section
// calling createDocumentChangeFeedQuery - so if we ever land in this if-block
// we know it is a terminal event.
CosmosException clientException = (CosmosException) throwable;
logger.warn("CosmosException: Partition {} from thread {} with owner {}",
this.lease.getLeaseToken(), Thread.currentThread().getId(), this.lease.getOwner(), clientException);
StatusCodeErrorType docDbError = ExceptionClassifier.classifyClientException(clientException);
switch (docDbError) {
case PARTITION_NOT_FOUND: {
this.resultException = new PartitionNotFoundException(
"Partition not found.",
this.lastServerContinuationToken);
}
break;
case PARTITION_SPLIT_OR_MERGE: {
this.resultException = new FeedRangeGoneException(
"Partition split.",
this.lastServerContinuationToken);
}
break;
case UNDEFINED: {
this.resultException = new RuntimeException(clientException);
}
break;
case MAX_ITEM_COUNT_TOO_LARGE: {
if (this.options.getMaxItemCount() <= 1) {
logger.error(
"Cannot reduce maxItemCount further as it's already at {}",
this.options.getMaxItemCount(),
clientException);
this.resultException = new RuntimeException(clientException);
}
this.options.setMaxItemCount(this.options.getMaxItemCount() / 2);
logger.warn("Reducing maxItemCount, new value: {}", this.options.getMaxItemCount());
return Flux.empty();
}
case TRANSIENT_ERROR: {
// Retry on transient (429) errors
if (clientException.getRetryAfterDuration().toMillis() > 0) {
Instant stopTimer = Instant.now().plus(clientException.getRetryAfterDuration().toMillis(), MILLIS);
return Mono.just(clientException.getRetryAfterDuration().toMillis()) // set some seed value to be able to run
// the repeat loop
.delayElement(Duration.ofMillis(100), CosmosSchedulers.COSMOS_PARALLEL)
.repeat(() -> {
Instant currentTime = Instant.now();
return !cancellationToken.isCancellationRequested() && currentTime.isBefore(stopTimer);
}).flatMap(values -> Flux.empty());
}
}
break;
default: {
logger.error("Unrecognized Cosmos exception returned error code {}", docDbError, clientException);
this.resultException = new RuntimeException(clientException);
}
}
} else if (throwable instanceof LeaseLostException) {
logger.info("LeaseLoseException with Partition {} from thread {} with owner {}",
this.lease.getLeaseToken(), Thread.currentThread().getId(), this.lease.getOwner());
this.resultException = (LeaseLostException) throwable;
} else if (throwable instanceof TaskCancelledException) {
logger.debug("Task cancelled exception: Partition {} from thread {} with owner {}",
this.lease.getLeaseToken(), Thread.currentThread().getId(), this.lease.getOwner(), throwable);
this.resultException = (TaskCancelledException) throwable;
} else {
logger.warn("Unexpected exception: Partition {} from thread {} with owner {}",
this.lease.getLeaseToken(), Thread.currentThread().getId(), this.lease.getOwner(), throwable);
this.resultException = new RuntimeException(throwable);
}
return Flux.error(throwable);
})
.repeat(() -> {
if (cancellationToken.isCancellationRequested()) {
this.resultException = new TaskCancelledException();
return false;
}
return true;
})
.onErrorResume(throwable -> {
if (this.resultException == null) {
this.resultException = new RuntimeException(throwable);
}
return Flux.empty();
})
.then()
.doFinally( any -> {
logger.info("Partition {}: processing task exited with owner {}.", this.lease.getLeaseToken(), this.lease.getOwner());
});
}