Method commitTransaction

src/CloudServices/CnchServerServiceImpl.cpp:314–369

void CnchServerServiceImpl::commitTransaction(
    google::protobuf::RpcController *  /*cntl*/,
    const Protos::CommitTransactionReq * request,
    Protos::CommitTransactionResp * response,
    google::protobuf::Closure * done)
{
    ContextPtr context_ptr = getContext();

    RPCHelpers::serviceHandler(
        done, response, [request = request, response = response, done = done, &global_context = *context_ptr, log = log] {
            brpc::ClosureGuard done_guard(done);
            response->set_commit_ts(0);

            try
            {
                /// Check validity of consumer before committing parts & offsets in case a new consumer has been scheduled
                if (request->has_kafka_storage_id())
                {
                    auto storage_id = RPCHelpers::createStorageID(request->kafka_storage_id());
                    auto bgthread = global_context.getCnchBGThread(CnchBGThreadType::Consumer, storage_id);
                    auto *manager = dynamic_cast<CnchKafkaConsumeManager *>(bgthread.get());

                    if (!manager->checkWorkerClient(storage_id.getTableName(), request->kafka_consumer_index()))
                        throw Exception(
                            "check validity of worker client for " + storage_id.getFullTableName() + " failed",
                            ErrorCodes::CNCH_KAFKA_TASK_NEED_STOP);
                    LOG_TRACE(
                        log,
                        "Check consumer {} OK. Now commit parts and offsets for Kafka transaction\n",storage_id.getFullTableName());
                }

                auto & txn_coordinator = global_context.getCnchTransactionCoordinator();
                auto txn_id = request->txn_id();
                auto txn = txn_coordinator.getTransaction(txn_id);

                // if (request->has_insertion_label())
                // {
                //     if (UUIDHelpers::Nil == txn->getMainTableUUID())
                //         throw Exception("Main table is not set when using insertion label", ErrorCodes::LOGICAL_ERROR);

                //     txn->setInsertionLabel(std::make_shared<InsertionLabel>(
                //         txn->getMainTableUUID(), request->insertion_label(), txn->getTransactionID().toUInt64()));
                // }

                auto commit_ts = txn->commit();
                response->set_commit_ts(commit_ts.toUInt64());

                LOG_TRACE(log, "Committed transaction from worker side: {}\n", request->txn_id());
            }
            catch (...)
            {
                tryLogCurrentException(log, __PRETTY_FUNCTION__);
                RPCHelpers::handleException(response->mutable_exception());
            }
        });
}

void CnchServerServiceImpl::precommitTransaction(