diff --git a/test/boost/aws_error_injection_test.cc b/test/boost/aws_error_injection_test.cc
index 739c67fa33..96f7972f28 100644
--- a/test/boost/aws_error_injection_test.cc
+++ b/test/boost/aws_error_injection_test.cc
@@ -20,11 +20,13 @@
 #include <seastar/util/closeable.hh>
 
 using namespace seastar;
-using namespace std::chrono_literals;
+using namespace std::string_view_literals;
+
 enum class failure_policy : uint8_t {
     SUCCESS = 0,
     RETRYABLE_FAILURE = 1,
     NONRETRYABLE_FAILURE = 2,
+    NEVERENDING_RETRYABLE_FAILURE = 3,
 };
 
 static uint16_t get_port() {
@@ -97,23 +99,34 @@ SEASTAR_THREAD_TEST_CASE(test_multipart_upload_file_retryable_success) {
     const size_t remainder_size = part_size / 2;
     const size_t total_size = 4 * part_size + remainder_size;
     const size_t memory_size = part_size;
-    BOOST_REQUIRE_EXCEPTION(test_client_upload_file(seastar_test::get_name(), failure_policy::RETRYABLE_FAILURE, total_size, memory_size), storage_io_error, [](const storage_io_error& e) {
-        return e.code().value() == EIO;
-    });
+    BOOST_REQUIRE_NO_THROW(test_client_upload_file(seastar_test::get_name(), failure_policy::RETRYABLE_FAILURE, total_size, memory_size));
 }
 
-SEASTAR_THREAD_TEST_CASE(test_multipart_upload_file_failure) {
+SEASTAR_THREAD_TEST_CASE(test_multipart_upload_file_failure_1) {
     const size_t part_size = 5_MiB;
     const size_t remainder_size = part_size / 2;
     const size_t total_size = 4 * part_size + remainder_size;
     const size_t memory_size = part_size;
-    BOOST_REQUIRE_EXCEPTION(test_client_upload_file(seastar_test::get_name(), failure_policy::NONRETRYABLE_FAILURE, total_size, memory_size),
-                            storage_io_error,
-                            [](const storage_io_error& e) { return e.code().value() == EIO; });
+    BOOST_REQUIRE_EXCEPTION(test_client_upload_file(seastar_test::get_name(), failure_policy::NEVERENDING_RETRYABLE_FAILURE, total_size, memory_size),
+            storage_io_error, [](const storage_io_error& e) {
+                return e.code().value() == EIO && e.what() == "S3 request failed. Reason: We encountered an internal error. Please try again."sv;
+            });
 }
 
-void do_test_client_multipart_upload(failure_policy policy) {
-    const sstring name(fmt::format("/{}/testobject-{}", "test", ::getpid()));
+SEASTAR_THREAD_TEST_CASE(test_multipart_upload_file_failure_2) {
+    const size_t part_size = 5_MiB;
+    const size_t remainder_size = part_size / 2;
+    const size_t total_size = 4 * part_size + remainder_size;
+    const size_t memory_size = part_size;
+    BOOST_REQUIRE_EXCEPTION(test_client_upload_file(seastar_test::get_name(), failure_policy::NONRETRYABLE_FAILURE, total_size, memory_size), storage_io_error,
+            [](const storage_io_error& e) {
+                return e.code().value() == EIO;
+            });
+}
+
+void do_test_client_multipart_upload(failure_policy policy, bool is_jumbo = false) {
+    const sstring name(fmt::format("/{}/testobject-{}-{}", "test", is_jumbo ? "jumbo" : "large", ::getpid()));
+
     register_policy(name, policy);
     testlog.info("Make client");
     semaphore mem(16 << 20);
@@ -121,7 +134,7 @@ void do_test_client_multipart_upload(failure_policy policy) {
     auto close_client = deferred_close(*cln);
 
     testlog.info("Upload object");
-    auto out = output_stream<char>(cln->make_upload_sink(name));
+    auto out = output_stream<char>(is_jumbo ? cln->make_upload_jumbo_sink(name, 3) : cln->make_upload_sink(name));
     auto close_stream = deferred_close(out);
 
     static constexpr unsigned chunk_size = 1000;
@@ -139,13 +152,38 @@ SEASTAR_THREAD_TEST_CASE(test_multipart_upload_sink_success) {
 }
 
 SEASTAR_THREAD_TEST_CASE(test_multipart_upload_sink_retryable_success) {
-    BOOST_REQUIRE_EXCEPTION(do_test_client_multipart_upload(failure_policy::RETRYABLE_FAILURE), storage_io_error, [](const storage_io_error& e) {
-        return e.code().value() == EIO;
+    BOOST_REQUIRE_NO_THROW(do_test_client_multipart_upload(failure_policy::RETRYABLE_FAILURE));
+}
+
+SEASTAR_THREAD_TEST_CASE(test_multipart_upload_sink_failure_1) {
+    BOOST_REQUIRE_EXCEPTION(do_test_client_multipart_upload(failure_policy::NEVERENDING_RETRYABLE_FAILURE), storage_io_error, [](const storage_io_error& e) {
+        return e.code().value() == EIO && e.what() == "S3 request failed. Reason: We encountered an internal error. Please try again."sv;
     });
 }
 
-SEASTAR_THREAD_TEST_CASE(test_multipart_upload_sink_failure) {
+SEASTAR_THREAD_TEST_CASE(test_multipart_upload_sink_failure_2) {
     BOOST_REQUIRE_EXCEPTION(do_test_client_multipart_upload(failure_policy::NONRETRYABLE_FAILURE), storage_io_error, [](const storage_io_error& e) {
-        return e.code().value() == EIO;
+        return e.code().value() == EIO && std::string_view(e.what()).contains("S3 request failed. Reason:");
+    });
+}
+
+SEASTAR_THREAD_TEST_CASE(test_multipart_upload_jumbo_sink_success) {
+    BOOST_REQUIRE_NO_THROW(do_test_client_multipart_upload(failure_policy::SUCCESS, true));
+}
+
+SEASTAR_THREAD_TEST_CASE(test_multipart_upload_jumbo_sink_retryable_success) {
+    BOOST_REQUIRE_NO_THROW(do_test_client_multipart_upload(failure_policy::RETRYABLE_FAILURE, true));
+}
+
+SEASTAR_THREAD_TEST_CASE(test_multipart_upload_jumbo_sink_failure_1) {
+    BOOST_REQUIRE_EXCEPTION(
+            do_test_client_multipart_upload(failure_policy::NEVERENDING_RETRYABLE_FAILURE, true), storage_io_error, [](const storage_io_error& e) {
+                return e.code().value() == EIO && e.what() == "S3 request failed. Reason: We encountered an internal error. Please try again."sv;
+            });
+}
+
+SEASTAR_THREAD_TEST_CASE(test_multipart_upload_jumbo_sink_failure_2) {
+    BOOST_REQUIRE_EXCEPTION(do_test_client_multipart_upload(failure_policy::NONRETRYABLE_FAILURE, true), storage_io_error, [](const storage_io_error& e) {
+        return e.code().value() == EIO && std::string_view(e.what()).contains("S3 request failed. Reason:");
     });
 }
diff --git a/test/pylib/s3_server_mock.py b/test/pylib/s3_server_mock.py
index 63e86ef236..9cbebfdb9e 100644
--- a/test/pylib/s3_server_mock.py
+++ b/test/pylib/s3_server_mock.py
@@ -25,6 +25,7 @@ class Policy(Enum):
     SUCCESS = 0
     RETRYABLE_FAILURE = 1
     NONRETRYABLE_FAILURE = 2
+    NEVERENDING_RETRYABLE_FAILURE = 3
 
 
 class LRUCache:
@@ -113,7 +114,6 @@ class InjectingHandler(BaseHTTPRequestHandler):
             put_data = self.rfile.read(int(content_length))
         self.send_response(200)
         self.send_header('Content-Type', 'text/plain; charset=utf-8')
-        self.send_header('Content-Length', '0')
         self.send_header('Connection', 'keep-alive')
         query_components = self.parsed_qs()
         if 'Key' in query_components and 'Policy' in query_components:
@@ -122,11 +122,28 @@ class InjectingHandler(BaseHTTPRequestHandler):
             self.send_header('ETag', "SomeTag_" + query_components.get("partNumber")[0])
         else:
             self.send_header('ETag', "SomeTag")
+
+        if self.headers['x-amz-copy-source']:
+            response_body = """<CopyPartResult>
+                                    <LastModified>2011-04-11T20:34:56.000Z</LastModified>
+                                    <ETag>"9b2cf535f27731c974343645a3985328"</ETag>
+                               </CopyPartResult>""".encode('utf-8')
+            self.send_header('Content-Length', str(len(response_body)))
+            self.end_headers()
+            self.wfile.write(response_body)
+            return
+
+        self.send_header('Content-Length', '0')
         self.end_headers()
 
     # Processes DELETE method, does nothing except providing in the response expected headers and response code
     def do_DELETE(self):
-        self.send_response(204)
+        query_components = self.parsed_qs()
+        if 'uploadId' in query_components and self.policies.get(
+                urlparse(self.path).path) == Policy.NONRETRYABLE_FAILURE:
+            self.send_response(404)
+        else:
+            self.send_response(204)
         self.send_header('Content-Type', 'text/plain; charset=utf-8')
         self.send_header('Content-Length', '0')
         self.send_header('Connection', 'keep-alive')
@@ -153,9 +170,10 @@ class InjectingHandler(BaseHTTPRequestHandler):
                                     <Key>Example-Object</Key>
                                     <ETag>"3858f62230ac3c915f300c664312c11f-9"</ETag>
                                 </CompleteMultipartUploadResult>"""
-                case Policy.RETRYABLE_FAILURE:
-                    # should succeed on retry
-                    self.policies.put(path, Policy.SUCCESS)
+                case Policy.RETRYABLE_FAILURE | Policy.NEVERENDING_RETRYABLE_FAILURE:
+                    if self.policies.get(path) == Policy.RETRYABLE_FAILURE:
+                        # should succeed on retry
+                        self.policies.put(path, Policy.SUCCESS)
                     return """<?xml version="1.0" encoding="UTF-8"?>
 
                                 <Error>