Method commit

src/Coordination/KeeperStateMachine.cpp:669–777

template<typename Storage>
nuraft::ptr<nuraft::buffer> KeeperStateMachine<Storage>::commit(const uint64_t log_idx, nuraft::buffer & data)
{
    const UInt64 start_time_us = ZooKeeperOpentelemetrySpans::now();

    auto request_for_session = parseRequest(data, true);
    if (!request_for_session->zxid)
        request_for_session->zxid = log_idx;

    request_for_session->log_idx = log_idx;

    if (!keeper_context->localLogsPreprocessed() && !preprocess(*request_for_session, /*lock_mutex=*/ true))
        return nullptr;

    const auto maybe_log_opentelemetry_span = [&](OpenTelemetry::SpanStatus status, const std::string & error_message)
    {
        request_for_session->request->spans.maybeInitialize(
            KeeperSpan::Commit,
            request_for_session->request->tracing_context.get(),
            start_time_us);

        request_for_session->request->spans.maybeFinalize(
            KeeperSpan::Commit,
            [&]
            {
                return std::vector<OpenTelemetry::SpanAttribute>{
                    {"keeper.operation", Coordination::opNumToString(request_for_session->request->getOpNum())},
                    {"keeper.session_id", request_for_session->session_id},
                    {"keeper.xid", request_for_session->request->xid},
                    {"raft.log_idx", log_idx},
                };
            },
            status,
            error_message);
    };

    try
    {
        const auto op_num = request_for_session->request->getOpNum();
        if (op_num == Coordination::OpNum::SessionID)
        {
            const Coordination::ZooKeeperSessionIDRequest & session_id_request
                = dynamic_cast<const Coordination::ZooKeeperSessionIDRequest &>(*request_for_session->request);
            int64_t session_id = 0;
            std::shared_ptr<Coordination::ZooKeeperSessionIDResponse> response = std::dynamic_pointer_cast<Coordination::ZooKeeperSessionIDResponse>(session_id_request.makeResponse());
            KeeperResponseForSession response_for_session;
            response_for_session.session_id = -1;
            response_for_session.response = response;
            response_for_session.request = request_for_session->request;

            KEEPER_STORAGE_LOCK_EXCLUSIVE(lock);
            session_id = storage->getSessionID(session_id_request.session_timeout_ms);
            LOG_DEBUG(log, "Session ID response {} with timeout {}", session_id, session_id_request.session_timeout_ms);
            response->session_id = session_id;
            if (response_callback)
                response_callback(std::move(response_for_session));
        }
        else
        {