diff options
Diffstat (limited to 'yt/cpp/mapreduce/client/client_reader.cpp')
| -rw-r--r-- | yt/cpp/mapreduce/client/client_reader.cpp | 17 |
1 files changed, 16 insertions, 1 deletions
diff --git a/yt/cpp/mapreduce/client/client_reader.cpp b/yt/cpp/mapreduce/client/client_reader.cpp index 1eade18397a..0bbc6fa1322 100644 --- a/yt/cpp/mapreduce/client/client_reader.cpp +++ b/yt/cpp/mapreduce/client/client_reader.cpp @@ -84,6 +84,11 @@ bool TClientReader::Retry( const TMaybe<ui64>& rowIndex, const std::exception_ptr& error) { + // We always stop retries if reader is aborted + if (IAbortableInputStream::IsAbortedError(error)) { + std::rethrow_exception(error); + } + if (CurrentRequestRetryPolicy_) { TMaybe<TDuration> backoffDuration; try { @@ -124,6 +129,16 @@ void TClientReader::ResetRetries() CurrentRequestRetryPolicy_ = nullptr; } +void TClientReader::Abort() +{ + Input_->Abort(); +} + +bool TClientReader::IsAborted() const +{ + return Input_->IsAborted(); +} + size_t TClientReader::DoRead(void* buf, size_t len) { return Input_->Read(buf, len); @@ -178,7 +193,7 @@ void TClientReader::CreateRequest(const TMaybe<ui32>& rangeIndex, const TMaybe<u ranges->begin()->LowerLimit(TReadLimit().RowIndex(*rowIndex)); } - Input_ = NDetail::RequestWithRetry<std::unique_ptr<IInputStream>>( + Input_ = NDetail::RequestWithRetry<std::unique_ptr<IAbortableInputStream>>( CurrentRequestRetryPolicy_, [this, &transactionId] (TMutationId /*mutationId*/) { return RawClient_->ReadTable(transactionId, Path_, Format_, Options_); |
