ElasticSearch Peer to Peer Recov
2018-12-04 本文已影响0人
persisting_
1 概述
ElasticSearch恢复（Recovery）分为几种模式，从
RecoverySource.Type中可以看到所有的模式：
//RecoverySource.Type
public enum Type {
EMPTY_STORE,
EXISTING_STORE,
//这种就是Peer to Peer恢复
PEER,
SNAPSHOT,
LOCAL_SHARDS
}
本文主要介绍Peer to Peer(P2P)恢复模式，在某个节点创建根据最新ClusterState创建完Shard之后，会进行恢复模式。
P2P恢复中涉及到Source端和Target端，P2P恢复是由Target发起，但是整个恢复流程又是由Source主导的。
Source端为恢复源所在的Node节点，Target端为需要从Source端接受数据并恢复到Source端状态的Node节点，恢复的结果为Target恢复为Source端数据。
2 预备知识
ElasticSearch在启动时会实例化一个Node实例，在Node初始化时，会分别初始化进行恢复必要的两个Service:
PeerRecoverySourceService以及PeerRecoveryTargetService。
其中PeerRecoverySourceService主要负责Source端的工作，PeerRecoveryTargetService则主要负责Target端的工作。
恢复过程涉及到的其他几个重要的类:
-
RecoverySourceHandler: Source端接受到Target端恢复请求时创建，在PeerRecoverySourceService负责主导恢复流程。 -
RemoteRecoveryTargetHandler: Source端RecoverySourceHandler持有的用于和Target交互（发送Segment快照、Translog等）的对象。 -
RecoveryTarget: Target端PeerRecoveryTargetService在启动恢复流程时创建的对象，用于接受Source端文件、Translog等执行恢复的对象。
此外，还需要注意一下PeerRecoverySourceService和PeerRecoveryTargetService在构造函数中向TransporgService注册的请求处理Handler。
Source端注册的Handler如下：
//PeerRecoverySourceService构造函数
//StartRecoveryTransportRequestHandler主要负责处理Target端启动恢复的
//请求，创建RecoverySourceHandler对象实例，开始恢复流程
transportService.registerRequestHandler(Actions.START_RECOVERY, StartRecoveryRequest::new, ThreadPool.Names.GENERIC, new StartRecoveryTransportRequestHandler());
Target端注册的Handler如下（后文介绍中可能有些没有涉及）：
//PeerRecoveryTargetService
//负责接受处理Source端Shard快照文件
transportService.registerRequestHandler(Actions.FILES_INFO, RecoveryFilesInfoRequest::new, ThreadPool.Names.GENERIC, new
FilesInfoRequestHandler());
//负责接受Source端发送的Shard文件块
transportService.registerRequestHandler(Actions.FILE_CHUNK, RecoveryFileChunkRequest::new, ThreadPool.Names.GENERIC, new
FileChunkTransportRequestHandler());
//负责将接受到的Shard文件更改为正式文件
transportService.registerRequestHandler(Actions.CLEAN_FILES, RecoveryCleanFilesRequest::new, ThreadPool.Names.GENERIC, new
CleanFilesRequestHandler());
//负责打开Engine，准备接受Source端Translog进行日志重放
transportService.registerRequestHandler(Actions.PREPARE_TRANSLOG, ThreadPool.Names.GENERIC,
RecoveryPrepareForTranslogOperationsRequest::new, new PrepareForTranslogOperationsRequestHandler());
//负责接受Source端Translog，进行日志重放
transportService.registerRequestHandler(Actions.TRANSLOG_OPS, RecoveryTranslogOperationsRequest::new, ThreadPool.Names.GENERIC,
new TranslogOperationsRequestHandler());
//负责刷新Engine，让新的Segment生效，回收旧的文件，更新global checkpoint等
transportService.registerRequestHandler(Actions.FINALIZE, RecoveryFinalizeRecoveryRequest::new, ThreadPool.Names.GENERIC, new
FinalizeRecoveryRequestHandler());
transportService.registerRequestHandler(Actions.WAIT_CLUSTERSTATE, RecoveryWaitForClusterStateRequest::new,
ThreadPool.Names.GENERIC, new WaitForClusterStateRequestHandler());
//如果是主分片relocate，则负责接受主分片身份
transportService.registerRequestHandler(
Actions.HANDOFF_PRIMARY_CONTEXT,
RecoveryHandoffPrimaryContextRequest::new,
ThreadPool.Names.GENERIC,
new HandoffPrimaryContextRequestHandler());
如上，Source和Target端注册的Handler负责处理指定阶段相互交互发送Request的请求，下文不再详述具体交互细节，会直接进入指定Handler进行介绍。
3 大致流程
P2P恢复大致流程.png
4 Target端发送开始恢复流程请求
Target端在createShade结束后会向Source端发送开始恢复流程请求。
//IndicesService
@Override
public IndexShard createShard(...) throws IOException {
ensureChangesAllowed();
IndexService indexService = indexService(shardRouting.index());
IndexShard indexShard = indexService.createShard(shardRouting, globalCheckpointSyncer);
indexShard.addShardFailureCallback(onShardFailure);
//Target端发起恢复流程起点
indexShard.startRecovery(recoveryState, recoveryTargetService, recoveryListener, repositoriesService,
(type, mapping) -> {
...
}, this);
return indexShard;
}
//IndexShard
public void startRecovery(...) {
...
switch (recoveryState.getRecoverySource().getType()) {
case EMPTY_STORE:
case EXISTING_STORE:
threadPool.generic().execute(() -> {
...
//从本地文件（Translog）恢复
if (recoverFromStore()) {
...
});
break;
case PEER:
...
//P2P恢复
recoveryTargetService.startRecovery(this, recoveryState.getSourceNode(), recoveryListener);
...
break;
case SNAPSHOT:
//从快照恢复
...
break;
case LOCAL_SHARDS:
//从本地Shard恢复
...
default:
throw new IllegalArgumentException("Unknown recovery source " + recoveryState.getRecoverySource());
}
}
//PeerRecoveryTargetService
//Target端PeerRecoveryTargetService开始恢复
public void startRecovery(final IndexShard indexShard, final DiscoveryNode sourceNode, final RecoveryListener listener) {
// create a new recovery status, and process...
//创建新的RecoveryTarget，并返回该RecoveryTarget对应的ID
//在发送START_RECOVERY给Source端时会同时发送该ID，后续
//Source端发送恢复流程中的request也会带上该ID，Target端根据
//该ID找到RecoveryTarget进行请求处理
final long recoveryId = onGoingRecoveries.startRecovery(indexShard, sourceNode, listener, recoverySettings.activityTimeout());
//Target具体恢复流程放在RecoveryRunner中
threadPool.generic().execute(new RecoveryRunner(recoveryId));
}
//PeerRecoveryTargetService.RecoveryRunner
class RecoveryRunner extends AbstractRunnable {
final long recoveryId;
RecoveryRunner(long recoveryId) {
this.recoveryId = recoveryId;
}
@Override
public void onFailure(Exception e) {
...
}
@Override
public void doRun() {
//具体的恢复函数
doRecovery(recoveryId);
}
}
//PeerRecoveryTargetService
private void doRecovery(final long recoveryId) {
...
try (
...
//生成request
request = getStartRecoveryRequest(recoveryTarget);
...
}
try {
cancellableThreads.execute(() -> responseHolder.set(
//发送START_RECOVERY到Source端，启动Source端恢复流程
transportService.submitRequest(request.sourceNode(), PeerRecoverySourceService.Actions.START_RECOVERY, request,
new FutureTransportResponseHandler<RecoveryResponse>() {
@Override
public RecoveryResponse newInstance() {
return new RecoveryResponse();
}
}).txGet()));
} catch (CancellableThreads.ExecutionCancelledException e) {
...
} catch (Exception e) {
//下面异常处理中某些异常可以重试恢复流程
Throwable cause = ExceptionsHelper.unwrapCause(e);
...
if (cause instanceof IllegalIndexShardStateException || cause instanceof IndexNotFoundException ||
..
retryRecovery(
..);
return;
}
if (cause instanceof DelayRecoveryException) {
retryRecovery(..);
return;
}
if (cause instanceof ConnectTransportException) {
..
retryRecovery(..);
return;
}
..
onGoingRecoveries.failRecovery(recoveryId, new RecoveryFailedException(request, e), true);
}
}
5 Source端开启恢复流程
5.1 RecoverySourceHandler的创建
Source端接收到Target发起的启动恢复流程START_RECOVERY请求后，使用StartRecoveryTransportRequestHandler处理该请求：
//PeerRecoverySourceService.StartRecoveryTransportRequestHandler
@Override
public void messageReceived(final StartRecoveryRequest request, final TransportChannel channel, Task task) throws Exception {
//Source端处理START_RECOVERY请求Handler调用
//PeerRecoverySourceService开启恢复流程
RecoveryResponse response = recover(request);
channel.sendResponse(response);
}
private RecoveryResponse recover(final StartRecoveryRequest request) throws IOException {
...
//这里创建RecoverySourceHandler用于主导恢复流程
RecoverySourceHandler handler = ongoingRecoveries.addNewRecovery(request, shard);
logger.trace("[{}][{}] starting recovery to {}", request.shardId().getIndex().getName(), request.shardId().id(), request.targetNode());
try {
//实际的恢复流程函数，Source端所有的恢复流程都是这个函数定义的，后续
//的恢复流程都是Source主导，发送各种请求让Target端进行文件恢复
//Translog重放等。
return handler.recoverToTarget();
} finally {
ongoingRecoveries.remove(shard, handler);
}
}
//PeerRecoverySourceService.OngoingRecoveries
synchronized RecoverySourceHandler addNewRecovery(StartRecoveryRequest request, IndexShard shard) {
final ShardRecoveryContext shardContext = ongoingRecoveries.computeIfAbsent(shard, s -> new ShardRecoveryContext());
//实际创建Handler的地方
RecoverySourceHandler handler = shardContext.addNewRecovery(request, shard);
shard.recoveryStats().incCurrentAsSource();
return handler;
}
//PeerRecoverySourceService.OngoingRecoveries.ShardRecoveryContext
synchronized RecoverySourceHandler addNewRecovery(StartRecoveryRequest request, IndexShard shard) {
for (RecoverySourceHandler existingHandler : recoveryHandlers) {
if (existingHandler.getRequest().targetAllocationId().equals(request.targetAllocationId())) {
throw new DelayRecoveryException("recovery with same target already registered, waiting for " +
"previous recovery attempt to be cancelled or completed");
}
}
RecoverySourceHandler handler = createRecoverySourceHandler(request, shard);
recoveryHandlers.add(handler);
return handler;
}
private RecoverySourceHandler createRecoverySourceHandler(StartRecoveryRequest request, IndexShard shard) {
RecoverySourceHandler handler;
//RecoverySourceHandler会持有一个RemoteRecoveryTargetHandler
//用于和Target端进行交互，后续各种请求都是通过这个对象进行交互的。
final RemoteRecoveryTargetHandler recoveryTarget =
new RemoteRecoveryTargetHandler(request.recoveryId(), request.shardId(), transportService,
request.targetNode(), recoverySettings, throttleTime -> shard.recoveryStats().addThrottleTime(throttleTime));
handler = new RecoverySourceHandler(shard, recoveryTarget, request, recoverySettings.getChunkSize().bytesAsInt());
return handler;
}
5.2 Source端主导流程recoverToTarget
Source端主导流程主要在RecoverySourceHandler.recoverToTarget函数中：
//RecoverySourceHandler
public RecoveryResponse recoverToTarget() throws IOException {
...
//创建Shard快照，该快照会在phase1阶段发送给Target端
phase1Snapshot = shard.acquireSafeIndexCommit();
...
phase1(phase1Snapshot.getIndexCommit(), () -> estimateNumOps);
...
prepareTargetForTranslog(isSequenceNumberBasedRecovery == false,
shard.estimateNumberOfHistoryOperations("peer-recovery", startingSeqNo));
...
targetLocalCheckpoint = phase2(startingSeqNo, requiredSeqNoRangeStart, endingSeqNo, snapshot,
maxSeenAutoIdTimestamp, maxSeqNoOfUpdatesOrDeletes);
...
finalizeRecovery(targetLocalCheckpoint);
...
return response;
}
通过上面的recoverToTarget函数可知，恢复流程主要分为另个阶段，额外加一个收尾阶段，分别为phase1,phase2,finalizeRecovery，下面依次介绍：
5.2.1 phase1：source端向Target发送快照
根据上述代码，在调用phase1函数之前，source端使用phase1Snapshot = shard.acquireSafeIndexCommit()对本地的数据执行快照操作，然后调用phase1发送快照文件到Target端。
phase1阶段主要向Target端发送三种请求:单个FILES_INFO告知Target即将接收的快照文件的相关信息、多个FILE_CHUNK进行文件传输、CLEAN_FILES告诉Target端清理原本存在但是恢复到Source端状态却不再需要的文件，CLEAN_FILES也会要求Target端对接收到的文件进行验证。
public void phase1(final IndexCommit snapshot, final Supplier<Integer> translogOps) {
//省略的内容主要是Target Start Recovery请求中的信息处理有差异的文件
...
//发送Shard文件相关信息到Target端
cancellableThreads.execute(() ->
recoveryTarget.receiveFileInfo(response.phase1FileNames, response.phase1FileSizes, response.phase1ExistingFileNames,
response.phase1ExistingFileSizes, translogOps.get()));
// How many bytes we've copied since we last called RateLimiter.pause
...
//通过文件块FILE_CHUNK的方式，将文件发送给Target端
sendFiles(store, phase1Files.toArray(new StoreFileMetaData[phase1Files.size()]), outputStreamFactories);
// Send the CLEAN_FILES request, which takes all of the files that
// were transferred and renames them from their temporary file
// names to the actual file names. It also writes checksums for
// the files after they have been renamed.
//
// Once the files have been renamed, any other files that are not
// related to this recovery (out of date segments, for example)
// are deleted
//发送CLEAN_FILES请求，告诉Target端临时文件中的文件命名为正式名称，删除旧的文件等。
try {
cancellableThreads.executeIO(() ->
recoveryTarget.cleanFiles(translogOps.get(), recoverySourceMetadata));
} catch (RemoteTransportException | IOException
...
}
其实Source端和Target端的交互基本上都是以请求和Handler的方式进行，比如Source端发送文件信息到Target端，即recoveryTarget.receiveFileInfo实现代码如下:
//RemoteRecoveryTargetHandler
@Override
public void receiveFileInfo(List<String> phase1FileNames, List<Long> phase1FileSizes, List<String> phase1ExistingFileNames,
List<Long> phase1ExistingFileSizes, int totalTranslogOps) {
//注意下面的recoveryId，Target端收到请求后会根据该ID
//找到负责助理本恢复的RecoveryTarget
RecoveryFilesInfoRequest recoveryInfoFilesRequest = new RecoveryFilesInfoRequest(recoveryId, shardId,
phase1FileNames, phase1FileSizes, phase1ExistingFileNames, phase1ExistingFileSizes, totalTranslogOps);
transportService.submitRequest(targetNode, PeerRecoveryTargetService.Actions.FILES_INFO, recoveryInfoFilesRequest,
TransportRequestOptions.builder().withTimeout(recoverySettings.internalActionTimeout()).build(),
EmptyTransportResponseHandler.INSTANCE_SAME).txGet();
}
具体如何发送的可以查看phase1函数代码，下面着重看下Target如何处理Source端发送的FILES_INFO、FILE_CHUNK、CLEAN_FILES请求：
Target端接收FILES_INFO请求后，会使用注册的FilesInfoRequestHandler处理：
//PeerRecoveryTargetService.FilesInfoRequestHandler
@Override
public void receiveFileInfo(List<String> phase1FileNames,
List<Long> phase1FileSizes,
List<String> phase1ExistingFileNames,
List<Long> phase1ExistingFileSizes,
int totalTranslogOps) {
final RecoveryState.Index index = state().getIndex();
//Target在此主要记录发送过来的快照文件的相关信息，如文件名、文件大小、事务日志等相关信息
for (int i = 0; i < phase1ExistingFileNames.size(); i++) {
index.addFileDetail(phase1ExistingFileNames.get(i), phase1ExistingFileSizes.get(i), true);
}
for (int i = 0; i < phase1FileNames.size(); i++) {
index.addFileDetail(phase1FileNames.get(i), phase1FileSizes.get(i), false);
}
state().getTranslog().totalOperations(totalTranslogOps);
state().getTranslog().totalOperationsOnStart(totalTranslogOps);
}
Source端发送完快照文件信息之后，会通过多个FILE_CHUNK请求向Target传输快
照文件，其实Target端在发送START_RECOVERY告诉Source端启动恢复流程时，也会将自己的元数据随着该请求发送给Source端，Source端在进行快照操作之后、发送给Target之前会依据自己的快照文件以及Target端请求中的元数据计算出差异的地方。
计算出的多个差异文件（即多个StoreFileMetaData）会以多个FILE_CHUNK请求发送给Target端。每个FILE_CHUNK携带一个完整的StoreFileMetaData。
Target端使用FileChunkTransportRequestHandler处理FILE_CHUNK请求,这里面需要注意一下ElasticSearch可使用参数参数indices.recovery.max_bytes_per_sec调节发送文件的速度：
//PeerRecoveryTargetService.FileChunkTransportRequestHandler
//Target端接收Source端发送过来的文件
@Override
public void messageReceived(final RecoveryFileChunkRequest request, TransportChannel channel, Task task) throws Exception {
try (RecoveryRef recoveryRef = onGoingRecoveries.getRecoverySafe(request.recoveryId(), request.shardId()
)) {
final RecoveryTarget recoveryTarget = recoveryRef.target();
final RecoveryState.Index indexState = recoveryTarget.state().getIndex();
if (request.sourceThrottleTimeInNanos() != RecoveryState.Index.UNKNOWN) {
indexState.addSourceThrottling(request.sourceThrottleTimeInNanos());
}
//这里默认使用Lucene的SimpleRateLimiter实现限速
RateLimiter rateLimiter = recoverySettings.rateLimiter();
if (rateLimiter != null) {
long bytes = bytesSinceLastPause.addAndGet(request.content().length());
if (bytes > rateLimiter.getMinPauseCheckBytes()) {
// Time to pause
bytesSinceLastPause.addAndGet(-bytes);
long throttleTimeInNanos = rateLimiter.pause(bytes);
indexState.addTargetThrottling(throttleTimeInNanos);
recoveryTarget.indexShard().recoveryStats().addThrottleTime(throttleTimeInNanos);
}
}
recoveryTarget.writeFileChunk(request.metadata(), request.position(), request.content(),
request.lastChunk(), request.totalTranslogOps()
);
}
channel.sendResponse(TransportResponse.Empty.INSTANCE);
}
Source端发送完所有差异文件之后，会发送CLEAN_FILES请求给Target端，Target端处理CLEAN_FILES的handler是：
//PeerRecoveryTargetService.CleanFilesRequestHandler
@Override
public void messageReceived(RecoveryCleanFilesRequest request, TransportChannel channel, Task task) throws Exception {
try (RecoveryRef recoveryRef = onGoingRecoveries.getRecoverySafe(request.recoveryId(), request.shardId()
)) {
recoveryRef.target().cleanFiles(request.totalTranslogOps(), request.sourceMetaSnapshot());
channel.sendResponse(TransportResponse.Empty.INSTANCE);
}
}
//RecoveryTarget
@Override
public void cleanFiles(int totalTranslogOps, Store.MetadataSnapshot sourceMetaData) throws IOException {
state().getTranslog().totalOperations(totalTranslogOps);
// first, we go and move files that were created with the recovery id suffix to
// the actual names, its ok if we have a corrupted index here, since we have replicas
// to recover from in case of a full cluster shutdown just when this code executes...
//将从Source端接收文件时创建的临时命名为正式文件，根据注释可知可能会覆盖Target端已有文件。
renameAllTempFiles();
final Store store = store();
store.incRef();
try {
//Target端清理本地原有文件，但是根据Source端发送过来恢复信息判断恢复之后不需要的文件，并对接收的文件进行验证
store.cleanupAndVerify("recovery CleanFilesRequestHandler", sourceMetaData);
if (indexShard.indexSettings().getIndexVersionCreated().before(Version.V_6_0_0_rc1)) {
store.ensureIndexHasHistoryUUID();
}
...
//其他相关处理
}
至此Source端phase1已经执行结束。Source端已经将本地快照文件发送给了Target端，Target端也已经对接收到的快照文件进行了处理，下面就需要进入Source端的phase2阶段，phase2端只要处理事务日志。
在进入phase2阶段之前，Source端还发送了PREPARE_TRANSLOG请求给Rarget端，告诉其打开Engine（ElasticSearch使用Engine执行写入操作），准备好接收Translog并重放。
Target端使用PrepareForTranslogOperationsRequestHandler处理PREPARE_TRANSLOG请求：
//PeerRecoveryTargetService.PrepareForTranslogOperationsRequestHandler
@Override
public void messageReceived(RecoveryPrepareForTranslogOperationsRequest request, TransportChannel channel,
Task task) throws Exception {
try (RecoveryRef recoveryRef = onGoingRecoveries.getRecoverySafe(request.recoveryId(), request.shardId()
)) {
recoveryRef.target().prepareForTranslogOperations(request.isFileBasedRecovery(), request.totalTranslogOps());
}
channel.sendResponse(TransportResponse.Empty.INSTANCE);
}
//RecoveryTarget
@Override
public void prepareForTranslogOperations(boolean fileBasedRecovery, int totalTranslogOps) throws IOException {
state().getTranslog().totalOperations(totalTranslogOps);
indexShard().openEngineAndSkipTranslogRecovery();
}
5.2.2 phase2：source端向Target发送Tanslog
Source端在phase2主要向Target发送事务日志：
//RecoverySourceHandler
long phase2(...)
...
// send all the snapshot's translog operations to the target
//调用RemoteRecoveryTargetHandler发送事务日志到Target端
//使用TRANSLOG_OPS请求
final SendSnapshotResult result = sendSnapshot(
startingSeqNo, requiredSeqNoRangeStart, endingSeqNo, snapshot, maxSeenAutoIdTimestamp, maxSeqNoOfUpdatesOrDeletes);
...
return result.targetLocalCheckpoint;
}
Target处理TRANSLOG_OPS请求的handler为TranslogOperationsRequestHandler:
//PeerRecoveryTargetService.TranslogOperationsRequestHandler
//这里不再展示TranslogOperationsRequestHandler源码，可以自行查看，//主要逻辑是Target端依次取出事务日志中的每个操作在本地进行重放。
5.2.3 收尾工作finalizeRecovery
Source端RecoverySourceHandler.recoverToTarget最后一步就是finalizeRecovery
//RecoverySourceHandler
/*
* finalizes the recovery process
*/
public void finalizeRecovery(final long targetLocalCheckpoint) throws IOException {
...
//忽略其他的逻辑，我们主要看下这里，如果Source端是主分片，则进行的是
//主分片重定位（PrimaryRelocation）
//恢复结束后需要移交Primary上下文给Target端
if (request.isPrimaryRelocation()) {
logger.trace("performing relocation hand-off");
// this acquires all IndexShard operation permits and will thus delay new recoveries until it is done
cancellableThreads.execute(() -> shard.relocated(recoveryTarget::handoffPrimaryContext));
/*
* if the recovery process fails after disabling primary mode on the source shard, both relocation source and
* target are failed (see {@link IndexShard#updateRoutingEntry}).
*/
}
stopWatch.stop();
logger.trace("finalizing recovery took [{}]", stopWatch.totalTime());
}
Source端移交Primary上下文主要通过发送HANDOFF_PRIMARY_CONTEXT报文：
@Override
public void handoffPrimaryContext(final ReplicationTracker.PrimaryContext primaryContext) {
transportService.submitRequest(
targetNode,
PeerRecoveryTargetService.Actions.HANDOFF_PRIMARY_CONTEXT,
new RecoveryHandoffPrimaryContextRequest(recoveryId, shardId, primaryContext),
TransportRequestOptions.builder().withTimeout(recoverySettings.internalActionTimeout()).build(),
EmptyTransportResponseHandler.INSTANCE_SAME).txGet();
}
Target端使用HandoffPrimaryContextRequestHandler处理HANDOFF_PRIMARY_CONTEXT报文:
//PeerRecoveryTargetService.HandoffPrimaryContextRequestHandler
@Override
public void messageReceived(final RecoveryHandoffPrimaryContextRequest request, final TransportChannel channel,
Task task) throws Exception {
try (RecoveryRef recoveryRef = onGoingRecoveries.getRecoverySafe(request.recoveryId(), request.shardId())) {
recoveryRef.target().handoffPrimaryContext(request.primaryContext());
}
channel.sendResponse(TransportResponse.Empty.INSTANCE);
}
//RecoveryTarget
@Override
public void handoffPrimaryContext(final ReplicationTracker.PrimaryContext primaryContext) {
indexShard.activateWithPrimaryContext(primaryContext);
}
Target端主要调用indexShard.activateWithPrimaryContext完成Primary上下文的接手，这里涉及到ReplicationTracker，比较复杂，后续会有专文介绍。
ReplicationTracker的介绍参考文章ElasticSearch ReplicationTracker
