[
https://issues.apache.org/jira/browse/SOLR-6463?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14139076#comment-14139076
]
Renaud Delbru commented on SOLR-6463:
-------------------------------------
Hi Yonik, All,
Here is a proposal for tracking update progress in CDCR. In this scenario, we
are assuming at the moment the Active-Passive scenario, where there is one
source cluster forwarding updates to one or more target clusters. Looking
forward to read your feedbacks on this proposal.
h4. Updates Tracking & Pushing
CDCR replicates data updates from the source to the target Data Center by
leveraging the Updates Log. A background thread regularly checks the Updates
Log for new entries, and then forwards them to the target Data Center. The
thread therefore needs to keep a checkpoint in the form of a pointer to the
last update successfully processed in the Updates Log. Upon acknowledgement
from the target Data Center that updates have been successfully processed, the
Updates Log pointer is updated to reflect the current checkpoint.
This pointer must be synchronized across all the replicas. In the case where
the leader goes down and a new leader is elected, the new leader will be able
to resume replication to the last update by using this synchronized pointer.
The strategy to synchronize such a pointer across replicas will be explained
next.
If for some reason, the target Data Center is offline or fails to process the
updates, the thread will periodically try to contact the target Data Center and
push the updates.
h4. Synchronization of Update Checkpoints
A reliable synchronization of the update checkpoints between the shard leader
and shard replicas is critical to avoid introducing inconsistency between the
source and target Data Centers. Another important requirement is that the
synchronization must be performed with minimal network traffic to maximize
scalability.
In order to achieve this, the strategy is to:
* Uniquely identify each update operation. This unique identifier will serve as
pointer.
* Rely on two storages: an ephemeral storage on the source shard leader, and a
persistent storage on the target cluster.
The shard leader in the source cluster will be in charge of generating a unique
identifier for each update operation, and will keep a copy of the identifier of
the last processed updates in memory. The identifier will be sent to the target
cluster as part of the update request. On the target Data Center side, the
shard leader will receive the update request, store it along with the unique
identifier in the Updates Log, and replicate it to the other shards.
SolrCloud is already providing a unique identifier for each update operation,
i.e., a “version” number. This version number is generated using a time-based
lamport clock which is incremented for each update operation sent. This
provides an “happened-before” ordering of the update operations that will be
leveraged in (1) the initialisation of the update checkpoint on the source
cluster, and in (2) the maintenance strategy of the Updates Log.
The persistent storage on the target cluster is used only during the election
of a new shard leader on the source cluster. If a shard leader goes down on the
source cluster and a new leader is elected, the new leader will contact the
target cluster to retrieve the last update checkpoint and instantiate its
ephemeral pointer. On such a request, the target cluster will retrieve the
latest identifier received across all the shards, and send it back to the
source cluster. To retrieve the latest identifier, every shard leader will look
up the identifier of the first entry in its Update Logs and sent it back to a
coordinator. The coordinator will have to select the highest among them.
This strategy does not require any additional network traffic and ensures
reliable pointer synchronization. Consistency is principally achieved by
leveraging SolrCloud. The update workflow of SolrCloud ensures that every
update is applied to the leader but also to any of the replicas. If the leader
goes down, a new leader is elected. During the leader election, a
synchronization is performed between the new leader and the other replicas. As
a result, this ensures that the new leader has a consistent Update Logs with
the previous leader. Having a consistent Updates Log means that:
* On the source cluster, the update checkpoint can be reused by the new leader.
* On the target cluster, the update checkpoint will be consistent between the
previous and new leader. This ensures the correctness of the update checkpoint
sent by a newly elected leader on the target cluster.
h6. Impact of Solr’s Update Reordering
The Updates Log can differ between the leader and the replicas, but not in an
inconsistent way. During leader to replica synchronisation, Solr’s Distributed
Update Processor will take care of reordering the update operations based on
their version number, and will drop any operations that are duplicate or could
cause inconsistency. One of the consequence is that the target cluster can send
back to the source cluster identifiers that do not exist anymore. However,
given that the identifier is an incremental version number, the update
checkpoint on the source cluster can be set to the next existing version number
without introducing inconsistency.
h6. Replication Between Clusters with Different Topology
The current design can work also in scenarios where replication is performed
between clusters with a different topology, e.g., one source cluster with two
shards and a target cluster with four shards. However, there is one limitation
due to a clock skew (version number) problem across shards.
In such a scenario, a target shard can receive updates from multiple source
shards (as the document ids will be redistributed across shards due to the
different cluster topology). This means that the version number generated by
the source cluster must be global to the cluster in order to keep partial
ordering of the updates. However, the version number is local to a shard. Given
that it is likely to have a clock skew across shards, a target shard will
receive updates with duplicate or non-ordered version numbers. This does not
really cause problems for add and delete-by-id operations, since the local
version number replicated to the target cluster will be able to keep partial
ordering for a given document identifier. However, this causes issues for
delete-by-query operations:
* When a cluster receives a delete-by-query, it is forwarded to each shard
leader.
* Each shard leader will assign a version number (which can end up being
different between shard leaders) to its delete-by-query, and replicate the
delete-by-query to all the target shard leaders.
* A target shard leader will receive more than one delete-by-query, with
possibly different version numbers. It will not be possible to duplicate and
reorder properly these delete-by-query.
One way to solve the problem of delete-by-query would be to have a clock
synchronisation procedure when a delete-by-query is received, which would
happen before the leader forwards the delete-by-query to the other leaders. The
workflow would look like the following:
* A leader receives a delete-by-query
* This (primary) leader requests a clock synchronisation across the cluster
(i.e., among the other leaders). The clock is synchronised by using the highest
version numbers across all the leaders.
* At this stage, the primary leader can assign a version number to the
delete-by-query, and forwards it to the other leaders.
* The secondary leaders does not overwrite and reuse the version number
attached to the delete-by-query.
The delete-by-query command will have the same version number across all the
leaders. When the leaders will replicate the commands to the target data
center, it then becomes possible to keep the partial ordering, since the source
leaders have been synchronised and the delete-by-query commands have all the
same version.
Therefore, the problem boils down on how to implement a clock synchronisation
procedure. Here is an initial proposal for a future option. Given that a
synchronisation will be done rarely (only in the case of a delete-by-query),
performance might be not critical for its implementation. A possible solution
would be a 2-phase communication approach, where the primary leader will
initiate the clock synch protocol, and will request the secondary leaders to:
* Block/buffer updates
* Send its latest version number to the primary leader
* Await the answer of the primary leader with the new clock
* Synchronise its clock
It is far from being perfect, as things might become tricky if there are
network or communication problems, but this is an initial idea to start
discussion.
> track update progress
> ---------------------
>
> Key: SOLR-6463
> URL: https://issues.apache.org/jira/browse/SOLR-6463
> Project: Solr
> Issue Type: Sub-task
> Reporter: Yonik Seeley
>
> Update progress needs to be tracked so that catch-up can be done after a
> network outage between clusters.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscr...@lucene.apache.org
For additional commands, e-mail: dev-h...@lucene.apache.org
