| 1208 | } |
| 1209 | |
| 1210 | void ClientRequestState::Finalize(const Status* cause) { |
| 1211 | if (otel_trace_query()) { |
| 1212 | // In a non-error case, end the query execution span since it will be the active span. |
| 1213 | if (cause == nullptr || cause->ok()) { |
| 1214 | otel_trace_manager_->EndChildSpanQueryExecution(); |
| 1215 | } |
| 1216 | |
| 1217 | // No need to end previous child span in an error case. This function silently closes |
| 1218 | // the active child span if there is one. |
| 1219 | otel_trace_manager_->StartChildSpanClose(cause); |
| 1220 | } |
| 1221 | |
| 1222 | Cancel(cause, /*wait_until_finalized=*/true); |
| 1223 | MarkActive(); |
| 1224 | // Make sure we join on wait_thread_ before we finish (and especially before this object |
| 1225 | // is destroyed). |
| 1226 | int64_t block_on_wait_time_us = 0; |
| 1227 | BlockOnWait(0, &block_on_wait_time_us); |
| 1228 | DCHECK_EQ(block_on_wait_time_us, 0); |
| 1229 | |
| 1230 | // Update latest observed Kudu timestamp stored in the session from the coordinator. |
| 1231 | // Needs to take the session_ lock which must not be taken while holding lock_, so this |
| 1232 | // must happen before taking lock_ below. |
| 1233 | Coordinator* coordinator = GetCoordinator(); |
| 1234 | if (coordinator != nullptr) { |
| 1235 | // This is safe to access on coord_ after Wait() has been called. |
| 1236 | uint64_t latest_kudu_ts = |
| 1237 | coordinator->dml_exec_state()->GetKuduLatestObservedTimestamp(); |
| 1238 | if (latest_kudu_ts > 0) { |
| 1239 | VLOG_RPC << "Updating session (id=" << PrintId(session_id()) << ") with latest " |
| 1240 | << "observed Kudu timestamp: " << latest_kudu_ts; |
| 1241 | lock_guard<mutex> session_lock(session_->lock); |
| 1242 | session_->kudu_latest_observed_ts = std::max<uint64_t>( |
| 1243 | session_->kudu_latest_observed_ts, latest_kudu_ts); |
| 1244 | } |
| 1245 | } |
| 1246 | |
| 1247 | // If the transaction didn't get committed by this point then we should just abort it. |
| 1248 | if (InTransaction()) { |
| 1249 | AbortTransaction(); |
| 1250 | } else if (InKuduTransaction()) { |
| 1251 | AbortKuduTransaction(); |
| 1252 | } |
| 1253 | |
| 1254 | UpdateEndTime(); |
| 1255 | |
| 1256 | { |
| 1257 | unique_lock<mutex> l(lock_); |
| 1258 | // Update result set cache metrics, and update mem limit accounting before tearing |
| 1259 | // down the coordinator. |
| 1260 | ClearResultCache(); |
| 1261 | } |
| 1262 | // Wait until the audit events are flushed. |
| 1263 | if (wait_thread_.get() != nullptr) { |
| 1264 | wait_thread_->Join(); |
| 1265 | wait_thread_.reset(); |
| 1266 | } else { |
| 1267 | // The query failed in the fe even before a wait thread is launched. Synchronously |
no test coverage detected