Partition export per part local backoff policy

docs/en/antalya/partition_export.md

@@ -61,11 +61,17 @@ TO TABLE [destination_database.]destination_table
 - **Default**: `false`
 - **Description**: Ignore existing partition export and overwrite the ZooKeeper entry. Allows re-exporting a partition that was already exported to the same destination. **IMPORTANT:** this is dangerous because it can lead to duplicated data, use it with caution.
 
-#### `export_merge_tree_partition_max_retries` (Optional)
+#### `export_merge_tree_partition_retry_initial_backoff_seconds` (Optional)
 
 - **Type**: `UInt64`
-- **Default**: `3`
-- **Description**: Maximum number of retries for exporting a merge tree part in an export partition task. If it exceeds, the entire task fails.
+- **Default**: `5`
+- **Description**: Initial delay (in seconds) before retrying a failed part export. The delay grows exponentially with the per-replica retry count (`delay = min(initial << (attempts - 1), max)`). The back-off is per-replica in-memory state: it only spaces this replica's retries out in time and never prevents another replica from attempting the same part. Retryable failures (transient memory/network/object-storage/Keeper errors) are retried until the task succeeds or `export_merge_tree_partition_task_timeout_seconds` elapses, while non-retryable failures (e.g. schema/type incompatibilities) fail the task immediately.
+
+#### `export_merge_tree_partition_retry_max_backoff_seconds` (Optional)
+
+- **Type**: `UInt64`
+- **Default**: `300`
+- **Description**: Maximum delay (in seconds) between retries of a failed part export. Caps the exponential growth controlled by `export_merge_tree_partition_retry_initial_backoff_seconds`.
 
 #### `export_merge_tree_part_file_already_exists_policy` (Optional)
 

src/Common/FailPoint.cpp

@@ -135,6 +135,8 @@ static struct InitFiu
     ONCE(iceberg_export_after_commit_before_zk_completed) \
     REGULAR(export_partition_commit_always_throw) \
     ONCE(export_partition_status_change_throw) \
+    REGULAR(export_part_non_retryable_throw) \
+    REGULAR(export_part_retryable_throw) \
     ONCE(backup_add_empty_memory_table) \
     PAUSEABLE_ONCE(backup_pause_on_start) \
     PAUSEABLE_ONCE(restore_pause_on_start) \

src/Core/Settings.cpp

@@ -7555,8 +7555,14 @@ Overwrite file if it already exists when exporting a merge tree part
     DECLARE(Bool, export_merge_tree_partition_force_export, false, R"(
 Ignore existing partition export and overwrite the zookeeper entry
 )", 0) \
-    DECLARE(UInt64, export_merge_tree_partition_max_retries, 3, R"(
-Maximum number of retries for exporting a merge tree part in an export partition task
+    DECLARE(UInt64, export_merge_tree_partition_retry_initial_backoff_seconds, 5, R"(
+Initial delay (in seconds) before retrying a failed part export in an export partition task.
+The delay grows exponentially with the per-replica retry count (capped doubling): `delay = min(initial << (attempts - 1), max)`, where `max` is `export_merge_tree_partition_retry_max_backoff_seconds`.
+The back-off is per-replica in-memory state: it only spaces this replica's retries out in time and never prevents another replica from attempting the same part. Retryable failures are retried until the task succeeds or `export_merge_tree_partition_task_timeout_seconds` elapses.
+To survive a long transient outage (e.g. object storage downtime), raise `export_merge_tree_partition_task_timeout_seconds`.
+)", 0) \
+    DECLARE(UInt64, export_merge_tree_partition_retry_max_backoff_seconds, 300, R"(
+Maximum delay (in seconds) between retries of a failed part export in an export partition task. Caps the exponential growth controlled by `export_merge_tree_partition_retry_initial_backoff_seconds`.
 )", 0) \
     DECLARE(UInt64, export_merge_tree_partition_task_timeout_seconds, 86400, R"(
 Maximum wall-clock duration (in seconds) an export partition task is allowed to remain in the PENDING state before it is auto-killed by the background cleanup loop.
@@ -7944,6 +7950,7 @@ Maximum number of WebAssembly UDF instances that can run in parallel per functio
 #define OBSOLETE_SETTINGS(M, ALIAS) \
     /** Obsolete settings which are kept around for compatibility reasons. They have no effect anymore. */ \
     MAKE_OBSOLETE(M, UInt64, export_merge_tree_partition_manifest_ttl, 86400) \
+    MAKE_OBSOLETE(M, UInt64, export_merge_tree_partition_max_retries, 3) \
     MAKE_OBSOLETE(M, Bool, query_condition_cache_store_conditions_as_plaintext, false) \
     MAKE_OBSOLETE(M, Bool, update_insert_deduplication_token_in_dependent_materialized_views, 0) \
     MAKE_OBSOLETE(M, UInt64, max_memory_usage_for_all_queries, 0) \

src/Core/SettingsChangesHistory.cpp

@@ -44,6 +44,9 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory()
             {"object_storage_cluster_join_mode", "allow", "allow", "New setting"},
             {"export_merge_tree_partition_task_timeout_seconds", "3600", "86400", "Increase default value to make it more realistic"},
             {"export_merge_tree_part_allow_lossy_cast", false, false, "New setting to gate lossy casts in EXPORT PART/PARTITION behind explicit acknowledgment"},
+            {"export_merge_tree_partition_retry_initial_backoff_seconds", 5, 5, "New setting for exponential back-off between failed part export retries in an export partition task"},
+            {"export_merge_tree_partition_retry_max_backoff_seconds", 300, 300, "New setting capping the exponential back-off between failed part export retries in an export partition task"},
+            {"export_merge_tree_partition_max_retries", 3, 3, "Obsolete and ignored: export partition tasks now retry retryable failures until the task timeout and fail immediately on non-retryable errors, instead of using a fixed retry budget"},
         });
         addSettingsChanges(settings_changes_history, "26.3",
         {
@@ -329,7 +332,6 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory()
             {"allow_experimental_export_merge_tree_part", false, true, "Turned ON by default for Antalya."},
             {"export_merge_tree_part_overwrite_file_if_exists", false, false, "New setting."},
             {"export_merge_tree_partition_force_export", false, false, "New setting."},
-            {"export_merge_tree_partition_max_retries", 3, 3, "New setting."},
             {"export_merge_tree_partition_manifest_ttl", 180, 180, "New setting."},
             {"export_merge_tree_part_file_already_exists_policy", "skip", "skip", "New setting."},
             {"hybrid_table_auto_cast_columns", true, true, "New setting to automatically cast Hybrid table columns when segments disagree on types. Default enabled."},

src/Storages/ExportReplicatedMergeTreePartitionManifest.h

@@ -23,7 +23,6 @@ struct ExportReplicatedMergeTreePartitionProcessingPartEntry
 
     String part_name;
     Status status;
-    size_t retry_count;
     String finished_by;
 
     std::string toJsonString() const
@@ -32,7 +31,6 @@ struct ExportReplicatedMergeTreePartitionProcessingPartEntry
 
         json.set("part_name", part_name);
         json.set("status", String(magic_enum::enum_name(status)));
-        json.set("retry_count", retry_count);
         json.set("finished_by", finished_by);
         std::ostringstream oss;     // STYLE_CHECK_ALLOW_STD_STRING_STREAM
         oss.exceptions(std::ios::failbit);
@@ -51,7 +49,6 @@ struct ExportReplicatedMergeTreePartitionProcessingPartEntry
 
         entry.part_name = json->getValue<String>("part_name");
         entry.status = magic_enum::enum_cast<Status>(json->getValue<String>("status")).value();
-        entry.retry_count = json->getValue<size_t>("retry_count");
         if (json->has("finished_by"))
         {
             entry.finished_by = json->getValue<String>("finished_by");
@@ -163,7 +160,8 @@ struct ExportReplicatedMergeTreePartitionManifest
     size_t number_of_parts;
     std::vector<String> parts;
     time_t create_time;
-    size_t max_retries;
+    size_t retry_initial_backoff_seconds = 5;
+    size_t retry_max_backoff_seconds = 300;
     size_t task_timeout_seconds;
     size_t max_threads;
     bool parallel_formatting;
@@ -208,7 +206,8 @@ struct ExportReplicatedMergeTreePartitionManifest
         json.set("file_already_exists_policy", String(magic_enum::enum_name(file_already_exists_policy)));
         json.set("filename_pattern", filename_pattern);
         json.set("create_time", create_time);
-        json.set("max_retries", max_retries);
+        json.set("retry_initial_backoff_seconds", retry_initial_backoff_seconds);
+        json.set("retry_max_backoff_seconds", retry_max_backoff_seconds);
         json.set("task_timeout_seconds", task_timeout_seconds);
         json.set("write_full_path_in_iceberg_metadata", write_full_path_in_iceberg_metadata);
         json.set("allow_lossy_cast", allow_lossy_cast);
@@ -236,7 +235,16 @@ struct ExportReplicatedMergeTreePartitionManifest
         manifest.destination_table = json->getValue<String>("destination_table");
         manifest.source_replica = json->getValue<String>("source_replica");
         manifest.number_of_parts = json->getValue<size_t>("number_of_parts");
-        manifest.max_retries = json->getValue<size_t>("max_retries");
+
+        if (json->has("retry_initial_backoff_seconds"))
+        {
+            manifest.retry_initial_backoff_seconds = json->getValue<size_t>("retry_initial_backoff_seconds");
+        }
+
+        if (json->has("retry_max_backoff_seconds"))
+        {
+            manifest.retry_max_backoff_seconds = json->getValue<size_t>("retry_max_backoff_seconds");
+        }
 
         if (json->has("iceberg_metadata_json"))
         {

src/Storages/MergeTree/ExportPartTask.cpp

@@ -18,6 +18,7 @@
 #include <QueryPipeline/QueryPipelineBuilder.h>
 #include "Common/setThreadName.h"
 #include <Common/Exception.h>
+#include <Common/FailPoint.h>
 #include <Common/ProfileEventsScope.h>
 #include <Databases/DatabaseReplicated.h>
 #include <Storages/MergeTree/ExportList.h>
@@ -43,6 +44,18 @@ namespace ErrorCodes
     extern const int FILE_ALREADY_EXISTS;
     extern const int LOGICAL_ERROR;
     extern const int QUERY_WAS_CANCELLED;
+    extern const int BAD_ARGUMENTS;
+    extern const int FAULT_INJECTED;
+}
+
+namespace FailPoints
+{
+    /// Throw a non-retryable (denylisted) error from the part-export worker, so the whole
+    /// export task transitions to FAILED immediately regardless of any timeout.
+    extern const char export_part_non_retryable_throw[];
+    /// Throw a retryable error from the part-export worker, so the part is retried with the
+    /// per-replica back-off until the task succeeds or the absolute timeout fires.
+    extern const char export_part_retryable_throw[];
 }
 
 namespace Setting
@@ -234,6 +247,18 @@ bool ExportPartTask::executeStep()
     {
         ThreadGroupSwitcher switcher((*exports_list_entry)->thread_group, ThreadName::EXPORT_PART);
 
+        fiu_do_on(FailPoints::export_part_non_retryable_throw,
+        {
+            throw Exception(ErrorCodes::BAD_ARGUMENTS,
+                "Failpoint: export_part_non_retryable_throw");
+        });
+
+        fiu_do_on(FailPoints::export_part_retryable_throw,
+        {
+            throw Exception(ErrorCodes::FAULT_INJECTED,
+                "Failpoint: export_part_retryable_throw");
+        });
+
         const auto filename = buildDestinationFilename(manifest, storage.getStorageID(), local_context);
 
         sink = destination_storage->import(

src/Storages/MergeTree/ExportPartitionManifestUpdatingTask.cpp

@@ -255,6 +255,8 @@ std::vector<ReplicatedPartitionExportInfo> ExportPartitionManifestUpdatingTask::
     if (!model)
         return {};
 
+    const auto backoff = storage.export_merge_tree_partition_task_scheduler->getLocalBackoffSnapshot();
+
     std::vector<ReplicatedPartitionExportInfo> infos;
     infos.reserve(model->size());
 
@@ -285,6 +287,13 @@ std::vector<ReplicatedPartitionExportInfo> ExportPartitionManifestUpdatingTask::
         }
         info.exception_count = total_exception_count;
 
+        if (const auto it = backoff.find(entry.getTransactionId()); it != backoff.end())
+        {
+            info.backoff_per_part.reserve(it->second.size());
+            for (const auto & [part_name, state] : it->second)
+                info.backoff_per_part.push_back({part_name, state.attempts, state.next_retry_time});
+        }
+
         infos.emplace_back(std::move(info));
     }
 
@@ -356,7 +365,20 @@ void ExportPartitionManifestUpdatingTask::poll()
                 continue;
             }
 
-            const auto metadata = ExportReplicatedMergeTreePartitionManifest::fromJsonString(metadata_json);
+            ExportReplicatedMergeTreePartitionManifest metadata;
+            try
+            {
+                metadata = ExportReplicatedMergeTreePartitionManifest::fromJsonString(metadata_json);
+            }
+            catch (...)
+            {
+                /// A single unparseable metadata.json (e.g. genuinely corrupt, or written by a
+                /// future incompatible format) must not abort the whole poll and stall discovery,
+                /// cleanup and status convergence for every other task. Skip just this entry.
+                tryLogCurrentException(storage.log, __PRETTY_FUNCTION__);
+                LOG_WARNING(storage.log, "ExportPartition Manifest Updating Task: Skipping {}: could not parse metadata.json", key);
+                continue;
+            }
 
             auto last_exception_per_replica = readLastExceptionPerReplica(
                 zk, fs::path(entry_path), key, storage.log.load());
@@ -490,20 +512,20 @@ void ExportPartitionManifestUpdatingTask::poll()
                 "Caught exception while committing export for {}: {}",
                 work.entry_path, e.message());
 
-            const bool exceeded_commimt_max_retries = ExportPartitionUtils::handleCommitFailure(
+            const bool became_failed = ExportPartitionUtils::handleCommitFailure(
                 zk,
                 work.entry_path,
-                work.metadata.max_retries,
+                e.code(),
                 storage.getReplicaName(),
                 e.message(),
                 log_ptr);
 
-            if (exceeded_commimt_max_retries)
+            if (became_failed)
             {
                 LOG_WARNING(log_ptr,
                     "ExportPartition Manifest Updating Task: "
-                    "Commit for {} transitioned to FAILED after exhausting max_retries={}",
-                    work.entry_path, work.metadata.max_retries);
+                    "Commit for {} transitioned to FAILED due to non-retryable error (code {})",
+                    work.entry_path, e.code());
             }
         }
     }