[1.x] Distinguish shard pick-up results

[ndolhii]

In this PR we start to distinguish the shard pick-up results. In particular, it is now possible to find out the reason of an unsuccessful shard pick-up. In particular, there may be some runtime issues, or the shard may already be picked-up by another worker.

Two new API endpoints were added to the DeliveryMonitor to provide end-users with some control over such cases:

1. FailedPickUp.Action onShardAlreadyPicked(AlreadyPickedUp failure)

This method will be invoked if the shard could not be picked as it's already picked by another worker. This method receives the ShardIndex of the shard that could not be picked, the WorkerId of the worker who owns the delivery session, and the Timestamp when the shard was picked. It is required to return an action to take in relation to this case. By default, it returns an action which just accepts this case (and ends the delivery session without any processing), but end-users may return a predefined action retrying the shard pick-up:

final class MyDeliveryMonitor extends DeliveryMonitor {

    ...

    @Override
    public FailedPickUp.Action onShardAlreadyPicked(AlreadyPickedUp failure) {
        return failure.retry();
    }

    ...

}

2. FailedPickUp.Action onShardPickUpFailure(RuntimeFailure failure)

This method is invoked if the shard could not be picked for some runtime technical reason, such as a runtime exception. This method receives the ShardIndex of the shard that could not be picked, and the instance of the occurred Exception. It also requires to return an action to handle this case. By default, such failures are just rethrown as RuntimeException, but end-users may choose to retry the pick-up:

final class MyDeliveryMonitor extends DeliveryMonitor {

    ...

    @Override
    public FailedPickUp.Action onShardPickUpFailure(RuntimeFailure failure) {
        return failure.retry();
    }

    ...

}

Breaking changes

The API of the ShardedWorkRegistry has been changed.

In particular, a new PickUpOutcome pickUp(ShardIndex index, NodeId node) method is introduced. Note, it returns an explicit result instead of Optional, as previously. This outcome contains either of two:

• a ShardSessionRecord — meaning that the shard is picked successfully,
• a ShardAlreadyPickedUp — a message that contains a WorkerID of the worker who owns the session at the moment, and the Timestamp when the shard was picked. This outcome means the session cannot be obtained as it's already picked.

Also, there is a new void release(ShardSessionRecord session) method that releases the passed session.

Here is a summary of code changes for those using ShardedWorkRegistry:

Before:

Optional<ShardProcessingSession> session = workRegistry.pickUp(index, currentNode);
if (session.isPresent()) { // Check if shard is picked.
   // ...
   session.get().complete(); // Release shard.
}

After:

PickUpOutcome outcome = workRegistry.pickUp(index, currentNode);
if (outcome.hasSession()) { // Check if shard is picked
    // ...
    workRegistry.release(outcome.getSession()); // Release shard.
}

Also, the new API allows getting the WorkerId of the worker who owns the session in case if the shard is already picked by someone else and the Timestamp when the shard was picked:

PickUpOutcome outcome = workRegistry.pickUp(index, currentNode);
if (outcome.hasAlreadyPickedBy()) {
    WorkerId worker = outcome.getAlreadyPicked().getWorker();
    Timestamp whenPicked = outcome.getAlreadyPicked().getWhenPicked();
    // ...
}

[armiol]

Please see my comments.

[armiol]

@nick-dolgiy please see my comments.

[ndolhii]

LGTM. A single question/suggestion: could the AlreadyPickedUp message include when_picked? It seems like a good spot to release a stale shard and retry the delivery.

Done.

[ndolhii]

@armiol PTAL.

[armiol]

@nick-dolgiy please see my comments.

[ndolhii]

@armiol PTAL.

[armiol]

@nick-dolgiy one last thing, hopefully. Please see my comment.

Otherwise, LGTM.

[armiol]

LGTM!