Skip to content

CosmosDb: Remove blocking calls from CFP #4602

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

Merged
christopheranderson merged 16 commits into from
Aug 20, 2019
Merged

CosmosDb: Remove blocking calls from CFP #4602

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
16 commits
Select commit Hold shift + click to select a range
0d876db
Remove blocking calls from ChangeFeedProcessor implementation and oth…
milismsft Jul 26, 2019
542566b
Merge branch 'master' into v3-cfp-part5
milismsft Jul 26, 2019
9d09623
Address PR feedback.
milismsft Aug 8, 2019
f228b6c
static logger
milismsft Aug 8, 2019
c428220
refactoring self->this
milismsft Aug 9, 2019
2cd06fc
Remove additional calls to Thread.sleep().
milismsft Aug 9, 2019
295c074
Remove redundant calls to Exceptions.propagate()
milismsft Aug 9, 2019
9a0aef5
Limit the feeds to one in order to avoid potential concurency issues …
milismsft Aug 14, 2019
0f4fceb
PR related refactoring.
milismsft Aug 14, 2019
d0eb617
PR related fix.
milismsft Aug 15, 2019
6d5cddf
PR related changes.
milismsft Aug 15, 2019
38fb38f
Add volatile to class members accessed from reactor code.
milismsft Aug 15, 2019
dc74c15
Add volatile to class members accessed from reactor code.
milismsft Aug 15, 2019
78aca21
Add volatile to class members accessed from reactor code.
milismsft Aug 15, 2019
ee4edf4
Replace usage of ExecutorService with Reactor's Scheduler (per conves…
milismsft Aug 15, 2019
24fe12d
The exception that results from the the child threads should be marke…
milismsft Aug 15, 2019
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
4 changes: 3 additions & 1 deletion ...re-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/ChangeFeedObserver.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.data.cosmos.internal.changefeed;

import com.azure.data.cosmos.CosmosItemProperties;
import reactor.core.publisher.Mono;

import java.util.List;

Expand Down Expand Up @@ -30,6 +31,7 @@ public interface ChangeFeedObserver {
*
* @param context the context specifying partition for this observer, etc.
* @param docs the documents changed.
* @return a deferred operation of this call.
*/
void processChanges(ChangeFeedObserverContext context, List<CosmosItemProperties> docs);
Mono<Void> processChanges(ChangeFeedObserverContext context, List<CosmosItemProperties> docs);
}
6 changes: 3 additions & 3 deletions ...src/main/java/com/azure/data/cosmos/internal/changefeed/exceptions/ObserverException.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,9 +11,9 @@ public class ObserverException extends RuntimeException {
/**
* Initializes a new instance of the {@link ObserverException} class using the specified internal exception.
*
* @param originalException {@link Exception} thrown by the user code.
* @param originalException {@link Throwable} thrown by the user code.
*/
public ObserverException(Exception originalException) {
super(DefaultMessage, originalException.getCause());
public ObserverException(Throwable originalException) {
super(DefaultMessage, originalException);
}
}
37 changes: 23 additions & 14 deletions .../main/java/com/azure/data/cosmos/internal/changefeed/implementation/AutoCheckpointer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
import com.azure.data.cosmos.internal.changefeed.ChangeFeedObserverCloseReason;
import com.azure.data.cosmos.internal.changefeed.ChangeFeedObserverContext;
import com.azure.data.cosmos.internal.changefeed.CheckpointFrequency;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.time.ZoneId;
Expand All @@ -19,12 +20,17 @@
class AutoCheckpointer implements ChangeFeedObserver {
private final CheckpointFrequency checkpointFrequency;
private final ChangeFeedObserver observer;
private int processedDocCount;
private ZonedDateTime lastCheckpointTime;
private volatile int processedDocCount;
private volatile ZonedDateTime lastCheckpointTime;

public AutoCheckpointer(CheckpointFrequency checkpointFrequency, ChangeFeedObserver observer) {
if (checkpointFrequency == null) throw new IllegalArgumentException("checkpointFrequency");
if (observer == null) throw new IllegalArgumentException("observer");
if (checkpointFrequency == null) {
throw new IllegalArgumentException("checkpointFrequency");
}

if (observer == null) {
throw new IllegalArgumentException("observer");
}

this.checkpointFrequency = checkpointFrequency;
this.observer = observer;
Expand All @@ -42,15 +48,22 @@ public void close(ChangeFeedObserverContext context, ChangeFeedObserverCloseReas
}

@Override
public void processChanges(ChangeFeedObserverContext context, List<CosmosItemProperties> docs) {
this.observer.processChanges(context, docs);
public Mono<Void> processChanges(ChangeFeedObserverContext context, List<CosmosItemProperties> docs) {
return this.observer.processChanges(context, docs)
.then(this.afterProcessChanges(context));
}

private Mono<Void> afterProcessChanges(ChangeFeedObserverContext context) {
this.processedDocCount ++;

if (this.isCheckpointNeeded()) {
context.checkpoint().block();
this.processedDocCount = 0;
this.lastCheckpointTime = ZonedDateTime.now(ZoneId.of("UTC"));
return context.checkpoint()
.doOnSuccess((Void) -> {
this.processedDocCount = 0;
this.lastCheckpointTime = ZonedDateTime.now(ZoneId.of("UTC"));
});
}
return Mono.empty();
}

private boolean isCheckpointNeeded() {
Expand All @@ -64,10 +77,6 @@ private boolean isCheckpointNeeded() {

Duration delta = Duration.between(this.lastCheckpointTime, ZonedDateTime.now(ZoneId.of("UTC")));

if (delta.compareTo(this.checkpointFrequency.getTimeInterval()) >= 0) {
return true;
}

return false;
return delta.compareTo(this.checkpointFrequency.getTimeInterval()) >= 0;
}
}
91 changes: 53 additions & 38 deletions .../main/java/com/azure/data/cosmos/internal/changefeed/implementation/BootstrapperImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,56 +21,71 @@ class BootstrapperImpl implements Bootstrapper {
private final Duration lockTime;
private final Duration sleepTime;

public BootstrapperImpl(PartitionSynchronizer synchronizer, LeaseStore leaseStore, Duration lockTime, Duration sleepTime)
{
if (synchronizer == null) throw new IllegalArgumentException("synchronizer");
if (leaseStore == null) throw new IllegalArgumentException("leaseStore");
if (lockTime == null || lockTime.isNegative() || lockTime.isZero()) throw new IllegalArgumentException("lockTime should be non-null and positive");
if (sleepTime == null || sleepTime.isNegative() || sleepTime.isZero()) throw new IllegalArgumentException("sleepTime should be non-null and positive");
private volatile boolean isInitialized;
private volatile boolean isLockAcquired;

public BootstrapperImpl(PartitionSynchronizer synchronizer, LeaseStore leaseStore, Duration lockTime, Duration sleepTime) {
if (synchronizer == null) {
throw new IllegalArgumentException("synchronizer");
}

if (leaseStore == null) {
throw new IllegalArgumentException("leaseStore");
}

if (lockTime == null || lockTime.isNegative() || lockTime.isZero()) {
throw new IllegalArgumentException("lockTime should be non-null and positive");
}

if (sleepTime == null || sleepTime.isNegative() || sleepTime.isZero()) {
throw new IllegalArgumentException("sleepTime should be non-null and positive");
}

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

this.isInitialized = false;
}

@Override
public Mono<Void> initialize() {
BootstrapperImpl self = this;
this.isInitialized = false;

return Mono.fromRunnable( () -> {
while (true) {
boolean initialized = self.leaseStore.isInitialized().block();
return Mono.just(this)
.flatMap( value -> this.leaseStore.isInitialized())
.flatMap(initialized -> {
this.isInitialized = initialized;

if (initialized) break;
if (initialized) {
return Mono.empty();
} else {
return this.leaseStore.acquireInitializationLock(this.lockTime)
.flatMap(lockAcquired -> {
this.isLockAcquired = lockAcquired;

boolean isLockAcquired = self.leaseStore.acquireInitializationLock(self.lockTime).block();

try {
if (!isLockAcquired) {
logger.info("Another instance is initializing the store");
try {
Thread.sleep(self.sleepTime.toMillis());
} catch (InterruptedException ex) {
logger.warn("Unexpected exception caught", ex);
}
continue;
}

logger.info("Initializing the store");
self.synchronizer.createMissingLeases().block();
self.leaseStore.markInitialized().block();

} catch (RuntimeException ex) {
break;
} finally {
if (isLockAcquired) {
self.leaseStore.releaseInitializationLock().block();
}
if (!this.isLockAcquired) {
logger.info("Another instance is initializing the store");
return Mono.just(isLockAcquired).delayElement(this.sleepTime);
} else {
return this.synchronizer.createMissingLeases()
.then(this.leaseStore.markInitialized());
}
})
.onErrorResume(throwable -> {
logger.warn("Unexpected exception caught", throwable);
return Mono.just(this.isLockAcquired);
})
.flatMap(lockAcquired -> {
if (this.isLockAcquired) {
return this.leaseStore.releaseInitializationLock();
}
return Mono.just(lockAcquired);
});
}
}

logger.info("The store is initialized");
});
})
.repeat( () -> !this.isInitialized)
.then();
}
}
10 changes: 6 additions & 4 deletions ...com/azure/data/cosmos/internal/changefeed/implementation/ChangeFeedContextClientImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,8 @@
import com.azure.data.cosmos.SqlQuerySpec;
import com.azure.data.cosmos.internal.PartitionKeyRange;
import com.azure.data.cosmos.internal.changefeed.ChangeFeedContextClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;
Expand All @@ -33,6 +35,8 @@
* Implementation for ChangeFeedDocumentClient.
*/
public class ChangeFeedContextClientImpl implements ChangeFeedContextClient {
private final Logger logger = LoggerFactory.getLogger(ChangeFeedContextClientImpl.class);

private final AsyncDocumentClient documentClient;
private final CosmosContainer cosmosContainer;
private Scheduler rxScheduler;
Expand All @@ -41,8 +45,7 @@ public class ChangeFeedContextClientImpl implements ChangeFeedContextClient {
* Initializes a new instance of the {@link ChangeFeedContextClient} interface.
* @param cosmosContainer existing client.
*/
public ChangeFeedContextClientImpl(CosmosContainer cosmosContainer)
{
public ChangeFeedContextClientImpl(CosmosContainer cosmosContainer) {
if (cosmosContainer == null) {
throw new IllegalArgumentException("cosmosContainer");
}
Expand All @@ -57,8 +60,7 @@ public ChangeFeedContextClientImpl(CosmosContainer cosmosContainer)
* @param cosmosContainer existing client.
* @param rxScheduler the RX Java scheduler to observe on.
*/
public ChangeFeedContextClientImpl(CosmosContainer cosmosContainer, Scheduler rxScheduler)
{
public ChangeFeedContextClientImpl(CosmosContainer cosmosContainer, Scheduler rxScheduler) {
if (cosmosContainer == null) {
throw new IllegalArgumentException("cosmosContainer");
}
Expand Down
3 changes: 1 addition & 2 deletions .../main/java/com/azure/data/cosmos/internal/changefeed/implementation/ChangeFeedHelper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -133,8 +133,7 @@ public static class KeyValuePair<K, V> implements Map.Entry<K, V>
private K key;
private V value;

public KeyValuePair(K key, V value)
{
public KeyValuePair(K key, V value) {
this.key = key;
this.value = value;
}
Expand Down
3 changes: 1 addition & 2 deletions ...m/azure/data/cosmos/internal/changefeed/implementation/ChangeFeedObserverContextImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,8 +24,7 @@ public ChangeFeedObserverContextImpl(String leaseToken) {
this.feedResponse = null;
}

public ChangeFeedObserverContextImpl(String leaseToken, FeedResponse<CosmosItemProperties> feedResponse, PartitionCheckpointer checkpointer)
{
public ChangeFeedObserverContextImpl(String leaseToken, FeedResponse<CosmosItemProperties> feedResponse, PartitionCheckpointer checkpointer) {
this.partitionKeyRangeId = leaseToken;
this.feedResponse = feedResponse;
this.checkpointer = checkpointer;
Expand Down
Loading