hcq/oceanbase
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

[OBCDC] fix a bug when fetch missing log time out

Browse Source
This commit is contained in:
obdev
2023-05-05 08:11:58 +00:00
committed by ob-robot
parent 6ec569ca8d
commit 6944111934
1 changed files with 25 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
src/logservice/cdcservice/ob_cdc_fetcher.cpp
View File

@ -839,6 +839,7 @@ int ObCdcFetcher::do_fetch_missing_log_(const obrpc::ObCdcLSFetchMissLogReq &req
resp.set_next_miss_lsn(missing_lsn); resp.set_next_miss_lsn(missing_lsn);
int64_t start_fetch_ts = ObTimeUtility::current_time(); int64_t start_fetch_ts = ObTimeUtility::current_time();
bool log_fetched_in_palf = false; bool log_fetched_in_palf = false;
bool log_fetched_in_archive = false;
if (is_time_up_(fetched_log_count, end_tstamp)) { // time up, stop fetching logs globally if (is_time_up_(fetched_log_count, end_tstamp)) { // time up, stop fetching logs globally
frt.stop("TimeUP"); frt.stop("TimeUP");
@ -863,32 +864,35 @@ int ObCdcFetcher::do_fetch_missing_log_(const obrpc::ObCdcLSFetchMissLogReq &req
if (OB_FAIL(fetch_log_in_archive_(ls_id, remote_iter, missing_lsn, if (OB_FAIL(fetch_log_in_archive_(ls_id, remote_iter, missing_lsn,
need_init_iter, log_entry, lsn, ctx))) { need_init_iter, log_entry, lsn, ctx))) {
LOG_WARN("fetch missng log in archive failed", KR(ret), K(missing_lsn)); LOG_WARN("fetch missng log in archive failed", KR(ret), K(missing_lsn));
} else {
log_fetched_in_archive = true;
} }
} }
}
if (OB_SUCC(ret)) {
if (OB_UNLIKELY(missing_lsn != lsn)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("do_fetch_missing_log missing_lsn not match", KR(ret), K(tenant_id_), K(ls_id),
K(missing_lsn), K(lsn));
} else {
resp.inc_log_fetch_time(ObTimeUtility::current_time() - start_fetch_ts);
check_next_entry_(lsn, log_entry, resp, frt);
if (frt.is_stopped()) { if (OB_SUCC(ret) && (log_fetched_in_palf || log_fetched_in_archive)) {
// Stop fetching log if (OB_UNLIKELY(missing_lsn != lsn)) {
} else if (OB_FAIL(prefill_resp_with_log_entry_(ls_id, lsn, log_entry, resp))) { ret = OB_ERR_UNEXPECTED;
if (OB_BUF_NOT_ENOUGH == ret) { LOG_WARN("do_fetch_missing_log missing_lsn not match", KR(ret), K(tenant_id_), K(ls_id),
handle_when_buffer_full_(frt); // stop K(missing_lsn), K(lsn));
ret = OB_SUCCESS;
} else {
LOG_WARN("prefill_resp_with_log_entry fail", KR(ret), K(frt), K(end_tstamp), K(resp));
}
} else { } else {
// log fetched successfully resp.inc_log_fetch_time(ObTimeUtility::current_time() - start_fetch_ts);
fetched_log_count++; check_next_entry_(lsn, log_entry, resp, frt);
LOG_TRACE("LS fetch a missing log", K(tenant_id_), K(ls_id), K(fetched_log_count), K(frt)); if (frt.is_stopped()) {
// Stop fetching log
} else if (OB_FAIL(prefill_resp_with_log_entry_(ls_id, lsn, log_entry, resp))) {
if (OB_BUF_NOT_ENOUGH == ret) {
handle_when_buffer_full_(frt); // stop
ret = OB_SUCCESS;
} else {
LOG_WARN("prefill_resp_with_log_entry fail", KR(ret), K(frt), K(end_tstamp), K(resp));
}
} else {
// log fetched successfully
fetched_log_count++;
LOG_TRACE("LS fetch a missing log", K(tenant_id_), K(ls_id), K(fetched_log_count), K(frt));
}
} }
} }
} }