public void finishCreateReplicaTask(CreateReplicaTask task, TFinishTaskRequest request) {
        long tabletId = task.getTabletId();
        TabletSchedCtx tabletCtx = takeRunningTablets(tabletId);
        if (tabletCtx == null) {
            LOG.warn("tablet info does not exist, tablet:{} backend:{}", tabletId, task.getBackendId());
            return;
        }

        // check if clone task success
        if (request.getTask_status().getStatus_code() != TStatusCode.OK) {
            LOG.warn("create replica task failed: {}", request.getTask_status().getError_msgs().get(0));
            finalizeTabletCtx(tabletCtx, TabletSchedCtx.State.CANCELLED, request.getTask_status().getError_msgs().get(0));
            return;
        }

        Replica replica = tabletCtx.getTablet().getReplicaByBackendId(task.getBackendId());
        if (replica == null) {
            LOG.warn("replica dose not exist, tablet:{} backend:{}", tabletId, task.getBackendId());
            finalizeTabletCtx(tabletCtx, TabletSchedCtx.State.CANCELLED, "replica not exist");
            return;
        }

        // write edit log
        replica.setState(ReplicaState.NORMAL);
        TabletMeta meta = GlobalStateMgr.getCurrentState().getTabletInvertedIndex().getTabletMeta(tabletId);
        ReplicaPersistInfo info = ReplicaPersistInfo.createForAdd(meta.getDbId(),
                meta.getTableId(), meta.getPhysicalPartitionId(), meta.getIndexId(),
                tabletId, replica.getBackendId(), replica.getId(), replica.getVersion(),
                replica.getSchemaHash(), replica.getDataSize(), replica.getRowCount(),
                replica.getLastFailedVersion(), replica.getLastSuccessVersion(),
                replica.getMinReadableVersion());
        GlobalStateMgr.getCurrentState().getEditLog().logAddReplica(info);
        finalizeTabletCtx(tabletCtx, TabletSchedCtx.State.FINISHED, "create replica finished");
        LOG.info("create replica for recovery successfully, tablet:{} backend:{}", tabletId, task.getBackendId());
    }
@Test
    public void testFinishCreateReplicaTask() {
        long beId = 10001L;
        long dbId = 10002L;
        long tblId = 10003L;
        long partitionId = 10004L;
        long indexId = 10005L;
        long tabletId = 10006L;
        long replicaId = 10007L;
        long schemaId = indexId;

        TTabletSchema tabletSchema = SchemaInfo.newBuilder().setId(schemaId)
                .setKeysType(DUP_KEYS)
                .setShortKeyColumnCount((short) 1)
                .setSchemaHash(-1)
                .setStorageType(TStorageType.COLUMN)
                .addColumn(new Column("k1", Type.INT))
                .build().toTabletSchema();

        CreateReplicaTask createReplicaTask = CreateReplicaTask.newBuilder()
                .setNodeId(beId)
                .setDbId(dbId)
                .setTableId(tblId)
                .setPartitionId(partitionId)
                .setIndexId(indexId)
                .setVersion(1)
                .setTabletId(tabletId)
                .setStorageMedium(TStorageMedium.HDD)
                .setPrimaryIndexCacheExpireSec(1)
                .setTabletType(TTabletType.TABLET_TYPE_DISK)
                .setCompressionType(TCompressionType.LZ4_FRAME)
                .setTabletSchema(tabletSchema)
                .build();

        TabletMeta tabletMeta = new TabletMeta(dbId, tblId, partitionId, indexId, -1, TStorageMedium.HDD);

        Replica replica = new Replica(replicaId, beId, -1, Replica.ReplicaState.RECOVER);

        tabletInvertedIndex.addTablet(tabletId, tabletMeta);
        tabletInvertedIndex.addReplica(tabletId, replica);

        TabletSchedCtx ctx = new TabletSchedCtx(TabletSchedCtx.Type.REPAIR,
                dbId, tblId, partitionId, indexId, tabletId, System.currentTimeMillis());
        LocalTablet tablet = new LocalTablet(tabletId);
        tablet.addReplica(replica);
        ctx.setTablet(tablet);

        TabletScheduler tabletScheduler = new TabletScheduler(new TabletSchedulerStat());

        TFinishTaskRequest request = new TFinishTaskRequest();
        TStatus status = new TStatus();
        status.setStatus_code(TStatusCode.OK);
        request.setTask_status(status);

        // failure test: running tablet ctx is not exist
        tabletScheduler.finishCreateReplicaTask(createReplicaTask, request);
        Assert.assertEquals(Replica.ReplicaState.RECOVER, replica.getState());

        // failure test: request not ok
        tabletScheduler.addToRunningTablets(ctx);
        status.setStatus_code(TStatusCode.CANCELLED);
        status.setError_msgs(Lists.newArrayList("canceled"));
        tabletScheduler.finishCreateReplicaTask(createReplicaTask, request);
        Assert.assertEquals(Replica.ReplicaState.RECOVER, replica.getState());

        // success
        tabletScheduler.addToRunningTablets(ctx);
        status.setStatus_code(TStatusCode.OK);
        tabletScheduler.finishCreateReplicaTask(createReplicaTask, request);
        Assert.assertEquals(Replica.ReplicaState.NORMAL, replica.getState());
    }