{

if (!client_config.retryStrategy)

throw Exception(ErrorCodes::LOGICAL_ERROR, "The S3 client can only be used with Client::RetryStrategy, define it in the client configuration");

assert_cast<const Client::RetryStrategy &>(*client_config.retryStrategy);

Aws::AmazonWebServiceRequest & request,

const HTTPHeaderEntries & extra_headers

)

{

for (const auto & [name, value] : extra_headers)

{

if (name.starts_with("x-amz-"))

{

request.SetAdditionalCustomHeaderValue(name, value);

}

}

{

if constexpr (IsReadMethod)

ProfileEvents::increment(read_event);

else

ProfileEvents::increment(write_event);

size_t max_redirects_,

ServerSideEncryptionKMSConfig sse_kms_config_,

const std::shared_ptr<Aws::Auth::AWSCredentialsProvider> & credentials_provider,

const PocoHTTPClientConfiguration & client_configuration,

Aws::Client::AWSAuthV4Signer::PayloadSigningPolicy sign_payloads,

const ClientSettings & client_settings)

{

verifyClientConfiguration(client_configuration);

return std::unique_ptr<Client>(

new Client(max_redirects_, std::move(sse_kms_config_), credentials_provider, client_configuration, sign_payloads, client_settings));

{

if (url.contains(".amazonaws.com"))

return ProviderType::AWS;

if (url.contains("storage.googleapis.com"))

return ProviderType::GCS;

return ProviderType::UNKNOWN;

size_t max_redirects_,

ServerSideEncryptionKMSConfig sse_kms_config_,

const std::shared_ptr<Aws::Auth::AWSCredentialsProvider> & credentials_provider_,

const PocoHTTPClientConfiguration & client_configuration_,

Aws::Client::AWSAuthV4Signer::PayloadSigningPolicy sign_payloads_,

const ClientSettings & client_settings_)

: Aws::S3::S3Client(credentials_provider_, client_configuration_, sign_payloads_, client_settings_.use_virtual_addressing)

, credentials_provider(credentials_provider_)

, client_configuration(client_configuration_)

, sign_payloads(sign_payloads_)

, client_settings(client_settings_)

, max_redirects(max_redirects_)

, sse_kms_config(std::move(sse_kms_config_))

, log(getLogger("S3Client"))

{

auto * endpoint_provider = dynamic_cast<Aws::S3::Endpoint::S3DefaultEpProviderBase *>(accessEndpointProvider().get());

endpoint_provider->GetBuiltInParameters().GetParameter("Region").GetString(explicit_region);

endpoint_provider->GetBuiltInParameters().GetParameter("Endpoint").GetString(initial_endpoint);

provider_type = deduceProviderType(initial_endpoint);

LOG_TRACE(log, "Provider type: {}", toString(provider_type));

LOG_TRACE(log, "Client configured with s3_retry_attempts: {}", client_configuration.retry_strategy.max_retries);

if (provider_type == ProviderType::GCS)

{

/// GCS can operate in 2 modes for header and query params names:

/// - with both x-amz and x-goog prefixes allowed (but cannot mix different prefixes in same request)

/// - only with x-goog prefix

/// first mode is allowed only with HMAC (or unsigned requests) so when we

/// find credential keys we can simply behave as the underlying storage is S3

/// otherwise, we need to be aware we are making requests to GCS

/// and replace all headers with a valid prefix when needed

if (credentials_provider)

{

auto credentials = credentials_provider->GetAWSCredentials();

if (credentials.IsEmpty())

api_mode = ApiMode::GCS;

}

}

LOG_TRACE(log, "API mode of the S3 client: {}", api_mode);

logConfiguration();

detect_region = provider_type == ProviderType::AWS && explicit_region == Aws::Region::AWS_GLOBAL;

cache = std::make_shared<ClientCache>();

ClientCacheRegistry::instance().registerClient(cache);

ProfileEvents::increment(ProfileEvents::S3Clients);

const Client & other, const PocoHTTPClientConfiguration & client_configuration_)

: Aws::S3::S3Client(other.credentials_provider, client_configuration_, other.sign_payloads,

other.client_settings.use_virtual_addressing)

, initial_endpoint(other.initial_endpoint)

, credentials_provider(other.credentials_provider)

, client_configuration(client_configuration_)

, sign_payloads(other.sign_payloads)

, client_settings(other.client_settings)

, explicit_region(other.explicit_region)

, detect_region(other.detect_region)

, provider_type(other.provider_type)

, api_mode(other.api_mode)

, max_redirects(other.max_redirects)

, sse_kms_config(other.sse_kms_config)

, log(getLogger("S3Client"))

{

cache = std::make_shared<ClientCache>(*other.cache);

ClientCacheRegistry::instance().registerClient(cache);

logConfiguration();

ProfileEvents::increment(ProfileEvents::TinyS3Clients);

{

if (detect_region)

return false;

if (error.GetResponseCode() == Aws::Http::HttpResponseCode::BAD_REQUEST && error.GetExceptionName() == "AuthorizationHeaderMalformed")

{

region = GetErrorMarshaller()->ExtractRegion(error);

if (region.empty())

region = getRegionForBucket(bucket, /*force_detect*/ true);

assert(!explicit_region.empty());

if (region == explicit_region)

return false;

insertRegionOverride(bucket, region);

return true;

}

return false;

{

std::lock_guard lock(cache->region_cache_mutex);

auto [it, inserted] = cache->region_for_bucket_cache.emplace(bucket, region);

if (inserted)

LOG_INFO(log, "Detected different region ('{}') for bucket {} than the one defined ('{}')", region, bucket, explicit_region);

{

// Don't do anything unless a key ID was specified

if (sse_kms_config.key_id)

{

request.SetServerSideEncryption(Model::ServerSideEncryption::aws_kms);

// If the key ID was specified but is empty, treat it as using the AWS managed key and omit the header

if (!sse_kms_config.key_id->empty())

request.SetSSEKMSKeyId(*sse_kms_config.key_id);

if (sse_kms_config.encryption_context)

request.SetSSEKMSEncryptionContext(*sse_kms_config.encryption_context);

if (sse_kms_config.bucket_key_enabled)

request.SetBucketKeyEnabled(*sse_kms_config.bucket_key_enabled);

}

{

const auto & bucket = request.GetBucket();

request.setApiMode(api_mode);

if (isS3ExpressBucket())

request.setIsS3ExpressBucket();

addAdditionalAMZHeadersToCanonicalHeadersList(request, client_configuration.extra_headers);

if (auto region = getRegionForBucket(bucket); !region.empty())

{

if (!detect_region)

LOG_INFO(log, "Using region override {} for bucket {}", region, bucket);

request.overrideRegion(std::move(region));

}

if (auto uri = getURIForBucket(bucket); uri.has_value())

request.overrideURI(std::move(*uri));

auto result = HeadObject(static_cast<const Model::HeadObjectRequest&>(request));

if (result.IsSuccess())

return result;

const auto & error = result.GetError();

std::string new_region;

if (checkIfWrongRegionDefined(bucket, error, new_region))

{

request.overrideRegion(new_region);

return Aws::S3::S3Client::HeadObject(request);

}

if (error.GetResponseCode() != Aws::Http::HttpResponseCode::MOVED_PERMANENTLY)

return result;

// maybe we detect a correct region

if (!detect_region)

{

if (auto region = GetErrorMarshaller()->ExtractRegion(error); !region.empty() && region != explicit_region)

{

request.overrideRegion(region);

insertRegionOverride(bucket, region);

}

}

auto bucket_uri = getURIForBucket(bucket);

if (!bucket_uri)

{

if (auto maybe_error = updateURIForBucketForHead(bucket); maybe_error.has_value())

return *maybe_error;

if (auto region = getRegionForBucket(bucket); !region.empty())

{

if (!detect_region)

LOG_INFO(log, "Using region override {} for bucket {}", region, bucket);

request.overrideRegion(std::move(region));

}

bucket_uri = getURIForBucket(bucket);

if (!bucket_uri)

{

LOG_ERROR(log, "Missing resolved URI for bucket {}, maybe the cache was cleaned", bucket);

return result;

}

}

const auto & current_uri_override = request.getURIOverride();

/// we already tried with this URI

if (current_uri_override && current_uri_override->uri == bucket_uri->uri)

{

LOG_INFO(log, "Getting redirected to the same invalid location {}", bucket_uri->uri.toString());

return result;

}

request.overrideURI(std::move(*bucket_uri));

/// The next call is NOT a recurcive call

/// This is a virtuall call Aws::S3::S3Client::HeadObject(const Model::HeadObjectRequest&)

return processRequestResult(

HeadObject(static_cast<const Model::HeadObjectRequest&>(request)));

{

return processRequestResult(

doRequest(request, [this](const Model::GetObjectTaggingRequest & req) { return GetObjectTagging(req); }));

{

return doRequestWithRetryNetworkErrors</*IsReadMethod*/ true>(

request, [this](Model::ListObjectsV2Request & req) { return ListObjectsV2(req); });

{

return doRequestWithRetryNetworkErrors</*IsReadMethod*/ true>(

request, [this](Model::ListObjectsRequest & req) { return ListObjects(req); });

{

return processRequestResult(

doRequest(request, [this](Model::GetObjectRequest & req) { return GetObject(req); }));

{

return doRequestWithRetryNetworkErrors</*IsReadMethod*/ false>(

request, [this](Model::AbortMultipartUploadRequest & req) { return AbortMultipartUpload(req); });

{

return doRequestWithRetryNetworkErrors</*IsReadMethod*/ false>(

request, [this](Model::CreateMultipartUploadRequest & req) { return CreateMultipartUpload(req); });

{

auto outcome = doRequestWithRetryNetworkErrors</*IsReadMethod*/ false>(

request, [this](Model::CompleteMultipartUploadRequest & req) { return CompleteMultipartUpload(req); });

const auto & key = request.GetKey();

const auto & bucket = request.GetBucket();

if (!outcome.IsSuccess()

&& outcome.GetError().GetErrorType() == Aws::S3::S3Errors::NO_SUCH_UPLOAD)

{

auto check_request = HeadObjectRequest()

.WithBucket(bucket)

.WithKey(key);

auto check_outcome = HeadObject(check_request);

/// if the key exists, than MultipartUpload has been completed at some of the retries

/// rewrite outcome with success status

if (check_outcome.IsSuccess())

outcome = Aws::S3::Model::CompleteMultipartUploadOutcome(Aws::S3::Model::CompleteMultipartUploadResult());

}

if (outcome.IsSuccess() && provider_type == ProviderType::GCS && client_settings.gcs_issue_compose_request)

{

/// For GCS we will try to compose object at the end, otherwise we cannot do a native copy

/// for the object (e.g. for backups)

/// We don't care if the compose fails, because the upload was still successful, only the

/// performance for copying the object will be affected

S3::ComposeObjectRequest compose_req;

compose_req.SetBucket(bucket);

compose_req.SetKey(key);

compose_req.SetComponentNames({key});

compose_req.SetContentType("binary/octet-stream");

auto compose_outcome = ComposeObject(compose_req);

if (compose_outcome.IsSuccess())

LOG_TRACE(log, "Composing object was successful");

else

LOG_INFO(

log,

"Failed to compose object. Message: {}, Key: {}, Bucket: {}",

compose_outcome.GetError().GetMessage(), key, bucket);

}

return outcome;

{

return doRequestWithRetryNetworkErrors</*IsReadMethod*/ false>(

request, [this](Model::CopyObjectRequest & req) { return CopyObject(req); });

{

return doRequestWithRetryNetworkErrors</*IsReadMethod*/ false>(

request, [this](Model::PutObjectRequest & req) { return PutObject(req); });

{

return doRequestWithRetryNetworkErrors</*IsReadMethod*/ false>(

request, [this](const Model::PutObjectTaggingRequest & req) { return PutObjectTagging(req); });

{

return doRequestWithRetryNetworkErrors</*IsReadMethod*/ false>(

request, [this](Model::UploadPartRequest & req) { return UploadPart(req); });

{

return doRequestWithRetryNetworkErrors</*IsReadMethod*/ false>(

request, [this](Model::UploadPartCopyRequest & req) { return UploadPartCopy(req); });

{

return doRequestWithRetryNetworkErrors</*IsReadMethod*/ false>(

request, [this](Model::DeleteObjectRequest & req) { Expect404ResponseScope scope; return DeleteObject(req); });

{

return doRequestWithRetryNetworkErrors</*IsReadMethod*/ false>(

request, [this](Model::DeleteObjectsRequest & req) { Expect404ResponseScope scope; return DeleteObjects(req); });

{

auto request_fn = [this](ComposeObjectRequest & req)

{

auto & endpoint_provider = const_cast<Client &>(*this).accessEndpointProvider();

AWS_OPERATION_CHECK_PTR(endpoint_provider, ComposeObject, Aws::Client::CoreErrors, Aws::Client::CoreErrors::ENDPOINT_RESOLUTION_FAILURE);

if (!req.BucketHasBeenSet())

{

AWS_LOGSTREAM_ERROR("ComposeObject", "Required field: Bucket, is not set")

return ComposeObjectOutcome(Aws::Client::AWSError<Aws::S3::S3Errors>(Aws::S3::S3Errors::MISSING_PARAMETER, "MISSING_PARAMETER", "Missing required field [Bucket]", false));

}

if (!req.KeyHasBeenSet())

{

AWS_LOGSTREAM_ERROR("ComposeObject", "Required field: Key, is not set")

return ComposeObjectOutcome(Aws::Client::AWSError<Aws::S3::S3Errors>(Aws::S3::S3Errors::MISSING_PARAMETER, "MISSING_PARAMETER", "Missing required field [Key]", false));

}

auto endpointResolutionOutcome = endpoint_provider->ResolveEndpoint(req.GetEndpointContextParams());

AWS_OPERATION_CHECK_SUCCESS(endpointResolutionOutcome, ComposeObject, Aws::Client::CoreErrors, Aws::Client::CoreErrors::ENDPOINT_RESOLUTION_FAILURE, endpointResolutionOutcome.GetError().GetMessage());

endpointResolutionOutcome.GetResult().AddPathSegments(req.GetKey());

endpointResolutionOutcome.GetResult().SetQueryString("?compose");

return ComposeObjectOutcome(MakeRequest(req, endpointResolutionOutcome.GetResult(), Aws::Http::HttpMethod::HTTP_PUT));

};

return doRequestWithRetryNetworkErrors</*IsReadMethod*/ false>(

request, request_fn);

{

addAdditionalAMZHeadersToCanonicalHeadersList(request, client_configuration.extra_headers);

const auto & bucket = request.GetBucket();

request.setApiMode(api_mode);

/// We have to use checksums for S3Express buckets, so the order of checks should be the following

if (client_settings.is_s3express_bucket)

request.setIsS3ExpressBucket();

else if (client_settings.disable_checksum)

request.disableChecksum();

if (auto region = getRegionForBucket(bucket); !region.empty())

{

if (!detect_region)

LOG_INFO(log, "Using region override {} for bucket {}", region, bucket);

request.overrideRegion(std::move(region));

}

if (auto uri = getURIForBucket(bucket); uri.has_value())

request.overrideURI(std::move(*uri));

bool found_new_endpoint = false;

// if we found correct endpoint after 301 responses, update the cache for future requests

SCOPE_EXIT(

if (found_new_endpoint)

{

auto uri_override = request.getURIOverride();

assert(uri_override.has_value());

updateURIForBucket(bucket, std::move(*uri_override));

}

);

for (size_t attempt = 0; attempt <= max_redirects; ++attempt)

{

auto result = request_fn(request);

if (result.IsSuccess())

return result;

const auto & error = result.GetError();

if (checkIfCredentialsChanged(error))

{

LOG_INFO(log, "Credentials changed, attempting again");

credentials_provider->SetNeedRefresh();

continue;

}

std::string new_region;

if (checkIfWrongRegionDefined(bucket, error, new_region))

{

request.overrideRegion(new_region);

continue;

}

/// IllegalLocationConstraintException may indicate that we are working with an opt-in region (e.g. me-south-1)

/// In that case, we need to update the region and try again

bool is_illegal_constraint_exception = error.GetExceptionName() == "IllegalLocationConstraintException";

if (error.GetResponseCode() != Aws::Http::HttpResponseCode::MOVED_PERMANENTLY && !is_illegal_constraint_exception)

return result;

// maybe we detect a correct region

if (!detect_region || is_illegal_constraint_exception)

{

if (auto region = GetErrorMarshaller()->ExtractRegion(error); !region.empty() && region != explicit_region)

{

LOG_INFO(log, "Detected new region: {}", region);

request.overrideRegion(region);

insertRegionOverride(bucket, region);

}

}

// we possibly got new location, need to try with that one

auto new_uri = is_illegal_constraint_exception ? std::optional<S3::URI>(initial_endpoint) : getURIFromError(error);

if (!new_uri)

return result;

if (initial_endpoint.substr(11) == "amazonaws.com") // Check if user didn't mention any region

new_uri->addRegionToURI(request.getRegionOverride());

const auto & current_uri_override = request.getURIOverride();

/// we already tried with this URI

if (current_uri_override && current_uri_override->uri == new_uri->uri)

{

LOG_INFO(log, "Getting redirected to the same invalid location {}", new_uri->uri.toString());

return result;

}

found_new_endpoint = true;

request.overrideURI(*new_uri);

}

throw Exception(ErrorCodes::TOO_MANY_REDIRECTS, "Too many redirects");

{

/// S3 does retries network errors actually.

/// But it does matter when errors occur.

/// This code retries a specific case when

/// network error happens when XML document is being read from the response body.

/// Hence, the response body is a stream, network errors are possible at reading.

/// S3 doesn't retry them.

/// Not all requests can be retried in that way.

/// Requests that read out response body to build the result are possible to retry.

/// Requests that expose the response stream as an answer are not retried with that code. E.g. GetObject.

addAdditionalAMZHeadersToCanonicalHeadersList(request, client_configuration.extra_headers);

auto with_retries = [this, request_fn_ = std::move(request_fn)] (RequestType & request_)

{

chassert(client_configuration.retryStrategy);

const Int64 max_attempts = client_configuration.retry_strategy.max_retries + 1;

Int64 attempt_no = 1;

std::invoke_result_t<RequestFn, RequestType &> outcome;

auto net_exception_handler = [&]() -> bool /// return true if we should retry

{

incrementProfileEvents<IsReadMethod>(ProfileEvents::S3ReadRequestsErrors, ProfileEvents::S3WriteRequestsErrors);

if (isClientForDisk())

incrementProfileEvents<IsReadMethod>(ProfileEvents::DiskS3ReadRequestsErrors, ProfileEvents::DiskS3WriteRequestsErrors);

tryLogCurrentException(log, fmt::format("Network error on S3 request, attempt {} of {}", attempt_no, max_attempts));

outcome = Aws::Client::AWSError<Aws::Client::CoreErrors>(

Aws::Client::CoreErrors::NETWORK_CONNECTION,

/*name*/ "",

/*message*/ fmt::format("All {} retry attempts failed. Last exception: {}", max_attempts, getCurrentExceptionMessage(false)),

/*retryable*/ true);

// network exceptions are always retryable, we could just return true here

// but we have to check cancellation points for query, ShouldRetry method does it already

return client_configuration.retryStrategy->ShouldRetry(outcome.GetError(), /*attemptedRetries*/ -1);

};

for (attempt_no = 1; attempt_no <= max_attempts; ++attempt_no)

{

incrementProfileEvents<IsReadMethod>(ProfileEvents::S3ReadRequestAttempts, ProfileEvents::S3WriteRequestAttempts);

if (isClientForDisk())

incrementProfileEvents<IsReadMethod>(ProfileEvents::DiskS3ReadRequestAttempts, ProfileEvents::DiskS3WriteRequestAttempts);

if (attempt_no > 1)

{

incrementProfileEvents<IsReadMethod>(ProfileEvents::S3ReadRequestRetryableErrors, ProfileEvents::S3WriteRequestRetryableErrors);

if (isClientForDisk())

incrementProfileEvents<IsReadMethod>(ProfileEvents::DiskS3ReadRequestRetryableErrors, ProfileEvents::DiskS3WriteRequestRetryableErrors);

// use previously attempt number to calculate delay

updateNextTimeToRetryAfterRetryableError(outcome.GetError(), attempt_no - 1);

// update ClickHouse-specific attempt number in the request

// to help choose the right timeouts on the HTTP client which depends on retry attempt number

auto clickhouse_request_attempt = getClickhouseAttemptNumber(request_);

setClickhouseAttemptNumber(request_, clickhouse_request_attempt + attempt_no);

}

/// Slowing down due to a previously encountered retryable error, possibly from another thread.

slowDownAfterRetryableError();

try

{

outcome = request_fn_(request_);

if (outcome.IsSuccess())

break;

// do not increment S3ReadRequestsErrors/S3WriteRequestsErrors here, it has been accounted in IO/S3/PocoHTTPClient.cpp

/// Retry attempts are managed by the outer loop, so the attemptedRetries argument can be ignored.

if (!client_configuration.retryStrategy->ShouldRetry(outcome.GetError(), /*attemptedRetries*/ -1))

break;

}

catch (Poco::Net::NetException &)

{

/// This includes "connection reset", "malformed message", and possibly other exceptions.

if (!net_exception_handler())

break;

}

catch (Poco::TimeoutException &)

{

/// This includes "Timeout"

if (!net_exception_handler())

break;

}

}

return outcome;

};

return doRequest(request, with_retries);

{

if (outcome.IsSuccess() || !isClientForDisk())

return std::forward<RequestResult>(outcome);

if (outcome.GetError().GetErrorType() == Aws::S3::S3Errors::NO_SUCH_KEY && !Expect404ResponseScope::is404Expected())

CurrentMetrics::add(CurrentMetrics::DiskS3NoSuchKeyErrors);

String enriched_message = fmt::format(

"{} {}",

outcome.GetError().GetMessage(),

Expect404ResponseScope::is404Expected() ? "This error is expected for S3 disk." : "This error happened for S3 disk.");

auto error = outcome.GetError();

error.SetMessage(enriched_message);

return RequestResult(error);

{

if (!client_configuration.s3_slow_all_threads_after_network_error && !client_configuration.s3_slow_all_threads_after_retryable_error)

return;

auto sleep_ms = client_configuration.retryStrategy->CalculateDelayBeforeNextRetry(error, attempt_no);

/// Other S3 requests must wait until this time.

UInt64 current_time_ms = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now().time_since_epoch()).count();

UInt64 next_time_ms = current_time_ms + sleep_ms;

UInt64 stored_next_time = next_time_to_retry_after_retryable_error;

/// Update to a later retry time, but only if it's further in the future.

while (stored_next_time < next_time_ms)

{

if (next_time_to_retry_after_retryable_error.compare_exchange_weak(stored_next_time, next_time_ms))

{

LOG_TRACE(log, "Updated next retry time to {} ms forward after retryable error with code {}", sleep_ms, error.GetResponseCode());

break;

}

}

{

if (!client_configuration.s3_slow_all_threads_after_network_error && !client_configuration.s3_slow_all_threads_after_retryable_error)

return;

/// Wait until `next_time_to_retry_after_retryable_error`.

for (;;)

{

UInt64 current_time_ms = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now().time_since_epoch()).count();

UInt64 next_time_ms = next_time_to_retry_after_retryable_error.load();

if (current_time_ms >= next_time_ms)

{

if (next_time_ms != 0)

LOG_TRACE(log, "Retry time has passed; proceeding without delay");

break;

}

UInt64 sleep_ms = next_time_ms - current_time_ms;

/// Adds jitter: a random factor in the range [100%, 110%] to the delay.

/// This prevents synchronized retries, reducing the risk of overwhelming the S3 server.

std::uniform_real_distribution<double> dist(1.0, 1.1);

double jitter = dist(thread_local_rng);

sleep_ms = static_cast<UInt64>(jitter * static_cast<double>(sleep_ms));

LOG_TRACE(log, "Request failed from a retryable error, now waiting {} ms before retrying", sleep_ms);

sleepForMilliseconds(sleep_ms);

}

{

if (client_configuration.for_disk_s3)

{

LOG_TRACE(

log,

"S3 client for disk '{}' initialized with s3_retry_attempts: {}",

client_configuration.opt_disk_name.value_or(""),

client_configuration.retry_strategy.max_retries);

LOG_TRACE(

log,

"S3 client for disk '{}': slowing down threads on retryable errors is {}",

client_configuration.opt_disk_name.value_or(""),

client_configuration.s3_slow_all_threads_after_retryable_error ? "enabled" : "disabled");

}

else

{

LOG_TRACE(log, "S3 client initialized with s3_retry_attempts: {}", client_configuration.retry_strategy.max_retries);

LOG_TRACE(

log,

"S3 client: slowing down threads on retryable errors is {}",

client_configuration.s3_slow_all_threads_after_retryable_error ? "enabled" : "disabled");

}

const std::shared_ptr<Aws::Http::HttpRequest>& httpRequest) const

{

Aws::S3::S3Client::BuildHttpRequest(request, httpRequest);

if (api_mode == ApiMode::GCS)

{

/// some GCS requests don't like S3 specific headers that the client sets

/// all "x-amz-*" headers have to be either converted or deleted

/// note that "amz-sdk-invocation-id" and "amz-sdk-request" are preserved

httpRequest->DeleteHeader("x-amz-api-version");

}

{

if (provider_type != ProviderType::GCS)

return "";

const auto & http_client = GetHttpClient();

if (!http_client)

return "";

auto * gcp_oauth_client = dynamic_cast<PocoHTTPClientGCPOAuth *>(http_client.get());

if (!gcp_oauth_client)

return "";

return gcp_oauth_client->getBearerToken();

{

std::lock_guard lock(cache->region_cache_mutex);

if (auto it = cache->region_for_bucket_cache.find(bucket); it != cache->region_for_bucket_cache.end())

return it->second;

if (!force_detect && !detect_region)

return "";

LOG_INFO(log, "Resolving region for bucket {}", bucket);

Aws::S3::Model::HeadBucketRequest req;

req.SetBucket(bucket);

addAdditionalAMZHeadersToCanonicalHeadersList(req, client_configuration.extra_headers);

std::string region;

auto outcome = HeadBucket(req);

if (outcome.IsSuccess())

{