#include <userver/kafka/exceptions.hpp>

#include <fmt/format.h>

USERVER_NAMESPACE_BEGIN

namespace kafka {

bool SendException::IsRetryable() const noexcept { return is_retryable_; }

SendException::SendException(const char* what, bool is_retryable)
    : std::runtime_error(what), is_retryable_(is_retryable) {}

DeliveryTimeoutException::DeliveryTimeoutException() : SendException(kWhat, /*is_retryable=*/true) {}

QueueFullException::QueueFullException() : SendException(kWhat, /*is_retryable=*/true) {}

MessageTooLargeException::MessageTooLargeException() : SendException(kWhat) {}

UnknownTopicException::UnknownTopicException() : SendException(kWhat) {}

UnknownPartitionException::UnknownPartitionException() : SendException(kWhat) {}

OffsetRangeException::OffsetRangeException(std::string_view what, std::string_view topic, std::uint32_t partition)
    : std::runtime_error(fmt::format("{} topic: '{}', partition: {}", what, topic, partition)) {}

OffsetRangeTimeoutException::OffsetRangeTimeoutException(std::string_view topic, std::uint32_t partition)
    : OffsetRangeException(kWhat, topic, partition) {}

GetMetadataException::GetMetadataException(std::string_view what, std::string_view topic)
    : std::runtime_error(fmt::format("{} topic: '{}'", what, topic)) {}

GetMetadataTimeoutException::GetMetadataTimeoutException(std::string_view topic) : GetMetadataException(kWhat, topic) {}

}  // namespace kafka

USERVER_NAMESPACE_END
