rethrow OrchestratorBlockedException

Author: bachuv

exporthistory/src/main/java/com/microsoft/durabletask/exporthistory/entity/ExportJob.java

@@ -177,6 +177,12 @@ private Object markAsFailed(TaskEntityOperation operation) {
         String jobId = operation.getContext().getId().getKey();
         String errorMessage = operation.getInput(String.class);
 
+        // Idempotent: a duplicate MARK_AS_FAILED (e.g. from replay) is a no-op rather than an error.
+        if (state.getStatus() == ExportJobStatus.FAILED) {
+            logger.fine("Export job already in FAILED state; ignoring duplicate MarkAsFailed: " + jobId);
+            return null;
+        }
+
         if (!ExportJobTransitions.isValidTransition(ExportJobOperationNames.MARK_AS_FAILED, state.getStatus(), ExportJobStatus.FAILED)) {
             throw new ExportJobInvalidTransitionException(
                     jobId, state.getStatus(), ExportJobStatus.FAILED, ExportJobOperationNames.MARK_AS_FAILED);

exporthistory/src/main/java/com/microsoft/durabletask/exporthistory/orchestrations/ExportJobOrchestrator.java

@@ -7,6 +7,8 @@
 import com.microsoft.durabletask.TaskOrchestration;
 import com.microsoft.durabletask.TaskOrchestrationContext;
 import com.microsoft.durabletask.RetryPolicy;
+import com.microsoft.durabletask.interruption.ContinueAsNewInterruption;
+import com.microsoft.durabletask.interruption.OrchestratorBlockedException;
 import com.microsoft.durabletask.exporthistory.constants.ExportJobOperationNames;
 import com.microsoft.durabletask.exporthistory.models.*;
 
@@ -125,6 +127,9 @@ public void run(TaskOrchestrationContext ctx) {
             // Mark completed
             ctx.callEntity(input.getJobEntityId(), ExportJobOperationNames.MARK_AS_COMPLETED).await();
 
+        } catch (OrchestratorBlockedException | ContinueAsNewInterruption controlFlow) {
+            // SDK control-flow signals — must propagate so the runtime can suspend/continue.
+            throw controlFlow;
         } catch (Exception ex) {
             // Mark as failed
             try {