jumpmindinc · gp510 · Aug 13, 2026 · Aug 13, 2026 · Pavel_JM (pavel-jm) · Aug 18, 2026
@@ -159,6 +159,8 @@ private ParameterConstants() {
     public static final String INITIAL_LOAD_CONCAT_CSV_IN_SQL_ENABLED = "initial.load.concat.csv.in.sql.enabled";
     public static final String INITIAL_LOAD_USE_COLUMN_TEMPLATES_ENABLED = "initial.load.use.column.templates.enabled";
     public static final String INITIAL_LOAD_EXTRACT_THREAD_COUNT_PER_SERVER = "initial.load.extract.thread.per.server.count";
+    public static final String INITIAL_LOAD_EXTRACT_REQUEST_RECOVERY_ENABLED = "initial.load.extract.request.recovery.enabled";
+    public static final String INITIAL_LOAD_EXTRACT_REQUEST_RECOVERY_THRESHOLD_MS = "initial.load.extract.request.recovery.threshold.ms";
     public static final String INITIAL_LOAD_EXTRACT_MAX_PROCESS_TIME_MS = "initial.load.extract.max.process.time.ms";
     public static final String INITIAL_LOAD_EXTRACT_TIMEOUT_MS = "initial.load.extract.timeout.ms";
     public static final String INITIAL_LOAD_EXTRACT_USE_TWO_PASS_LOB = "initial.load.extract.use.two.pass.lob";

@@ -67,6 +67,15 @@ public ExtractRequest requestExtractRequest(ISqlTransaction transaction, String
 
     public void resetExtractRequest(OutgoingBatch batch);
 
+    /**
+     * Return extract requests to NE when they are marked complete but their batches are still requested, which is the state a load is left in when the JVM is
+     * interrupted mid-extract. Called automatically from {@link #queueWork(boolean)}; exposed so an operator can drive it directly, and so a request whose
+     * range contains already-delivered batches can be recovered with {@code force} once the target has been dealt with.
+     *
+     * @return the number of requests restarted
+     */
+    public int recoverStuckExtractRequests(boolean force);
+
     public void removeBatchFromStaging(OutgoingBatch batch);
 
     public StagingFileLock acquireStagingFileLock(OutgoingBatch batch);

@@ -1050,9 +1050,22 @@
                     }
                     ExtractRequest extractRequest = getExtractRequestForBatch(currentBatch);
                     if (extractRequest != null && extractRequest.getStatus() != ExtractStatus.OK) {
-                        sqlTemplate.update(getSql("updateExtractRequestStatus"), ExtractStatus.OK.name(), new Date(),
-                                currentBatch.getExtractRowCount(), currentBatch.getExtractMillis(), extractRequest.getRequestId());
-                        checkSendDeferredForeignKeys(extractRequest.getLoadId(), targetNode);
+                        if (isExtractRequestComplete(extractRequest, currentBatch, mode)) {
+                            sqlTemplate.update(getSql("updateExtractRequestStatus"), ExtractStatus.OK.name(), new Date(),
+                                    currentBatch.getExtractRowCount(), currentBatch.getExtractMillis(), extractRequest.getRequestId());
+                            checkSendDeferredForeignKeys(extractRequest.getLoadId(), targetNode);
+                        } else {
+                            /*
+                             * One batch finishing does not mean the request did. Marking the whole request OK from a single batch's counters is how a request
+                             * ends up at OK with extracted_rows=0 while its batches are still RQ: every extraction selector filters on NE so nothing
+                             * re-extracts it, extractBatch refuses to deliver an RQ batch, and nothing reconciles the two or raises an error. Accumulate the
+                             * statistics instead, so progress stays visible without claiming completion.
+                             */
+                            sqlTemplate.update(getSql("updateExtractRequestExtractedStats"), currentBatch.getExtractRowCount(),
+                                    currentBatch.getExtractMillis(), new Date(), extractRequest.getRequestId());
+                            log.debug("Batch {} finished but extract request {} for table {} is not complete; leaving its status at {}",
+                                    currentBatch.getBatchId(), extractRequest.getRequestId(), extractRequest.getTableName(), extractRequest.getStatus());
+                        }
                     }
                 }
             }
@@ -1720,13 +1733,19 @@
         return transformExtractWriter;
     }
 
+    /** When recovery last scanned for stuck extract requests, so it is not re-run faster than a request can become stale. */
+    protected long lastStuckExtractRequestCheckMs;
+
     @Override
     public RemoteNodeStatuses queueWork(boolean force) {
         final RemoteNodeStatuses statuses = new RemoteNodeStatuses(configurationService.getChannels(false));
         Node identity = nodeService.findIdentity();
         if (identity != null) {
             if (force || clusterService.lock(ClusterConstants.INITIAL_LOAD_EXTRACT)) {
                 try {
+                    // Before assigning threads, put back any request that is marked complete but demonstrably is not.
+                    // The NE requests this produces are picked up by getExtractRequestNodes() in this same invocation.
+                    recoverStuckExtractRequests(false);
                     updateExtractRequestsForThreading();
                     List<NodeQueuePair> nodes = getExtractRequestNodes();
                     for (NodeQueuePair pair : nodes) {
@@ -2087,6 +2106,106 @@
         }
     }
 
+    /**
+     * Whether finishing this batch means the whole extract request is done.
+     * <p>
+     * The batch-status half of the test is exact rather than heuristic: {@code MultiBatchStagingWriter.close()} advances every remaining batch in the range, so
+     * after a completed extract-job run none of them are left at {@code RQ}. A request at {@code OK} with any batch in its range still {@code RQ} is therefore
+     * an impossible state rather than a healthy in-flight one, which is what makes both this guard and {@link #recoverStuckExtractRequests(boolean)} safe from
+     * false positives. The zero-row check is a heuristic and is called out as such where it appears.
+     */
+    protected boolean isExtractRequestComplete(ExtractRequest request, OutgoingBatch currentBatch, ExtractMode mode) {
+        if (mode == ExtractMode.EXTRACT_ONLY) {
+            // changeBatchStatus does not persist in this mode, so the batch's own row would still read RQ.
+            return false;
+        }
+        if (request.getEndBatchId() > request.getStartBatchId() && currentBatch.getExtractRowCount() == 0 && request.getRows() > 0) {
+            /*
+             * Heuristic, and checked before the query because it costs nothing: a multi-batch request whose expected row count is non-zero cannot be finished
+             * by a batch that extracted nothing, because those rows belong to another batch in the range. A legitimately empty final batch on a request that
+             * expects rows is the case this misjudges, and it errs towards leaving the request open, which the reconciler then resolves.
+             */
+            return false;
+        }
+        return !hasRequestedBatchesInRange(request);
+    }
+
+    /**
+     * Whether any batch in the request's range is still requested. Deliberately not a {@code count(*)}: the answer is only ever used as a boolean, and a count
+     * has to visit every batch in the range, so across a load split into thousands of batches the cost would grow with the square of the batch count. Fetching
+     * one row stops at the first match, which in the common in-progress case is immediate.
+     */
+    protected boolean hasRequestedBatchesInRange(ExtractRequest request) {
+        return !sqlTemplate.query(getSql("selectRequestedBatchesForExtractRequestSql"), 1, new LongMapper(),
+                new Object[] { request.getNodeId(), request.getStartBatchId(), request.getEndBatchId() }, null).isEmpty();
+    }
+
+    /** Whether any batch in the request's range was already delivered, which makes an automatic restart unsafe. */
+    protected boolean hasDeliveredBatchesInRange(ExtractRequest request) {
+        return !sqlTemplate.query(getSql("selectDeliveredBatchesForExtractRequestSql"), 1, new LongMapper(),
+                new Object[] { request.getNodeId(), request.getStartBatchId(), request.getEndBatchId() }, null).isEmpty();
+    }
+
+    /**
+     * Return extract requests to {@code NE} when they are marked complete but demonstrably are not, so a load interrupted mid-extract resumes instead of
+     * sitting silently forever. Runs from {@link #queueWork(boolean)}, which also covers startup; a startup-only check would not have helped the reporting
+     * site, whose node ran three days in this state.
+     * <p>
+     * Requests whose range contains already-delivered batches are <em>not</em> restarted automatically. {@code restartExtractRequest} flips the whole range
+     * back to {@code RQ} through a statement with no status predicate, so it would re-send rows already committed at the target, and the target's own record of
+     * them can be missing (that is the data-integrity half of this defect). Those are reported and require {@code force}.
+     *
+     * @return the number of requests restarted
+     */
+    @Override
+    public int recoverStuckExtractRequests(boolean force) {
+        if (!parameterService.is(ParameterConstants.INITIAL_LOAD_EXTRACT_REQUEST_RECOVERY_ENABLED, true)) {
+            return 0;
+        }
+        long thresholdMs = parameterService.getLong(ParameterConstants.INITIAL_LOAD_EXTRACT_REQUEST_RECOVERY_THRESHOLD_MS, 300000);
+        long now = System.currentTimeMillis();
+        /*
+         * A request cannot qualify until it has been untouched for the threshold, so scanning faster than that only re-runs a query that cannot have a new
+         * answer. queueWork runs every 10 seconds; at the default threshold that would be 29 wasted scans out of every 30, each walking every completed request
+         * for this node plus a correlated probe into the outgoing batch table.
+         */
+        if (!force && now - lastStuckExtractRequestCheckMs < thresholdMs) {
+            return 0;
+        }
+        lastStuckExtractRequestCheckMs = now;
+        List<ExtractRequest> stuck = sqlTemplateDirty.query(getSql("selectStuckExtractRequestsSql"), new ExtractRequestMapper(),
+                engine.getNodeId(), ExtractStatus.OK.name(), new Date(now - thresholdMs));
+        int restarted = 0;
+        for (ExtractRequest request : stuck) {
+            if (hasDeliveredBatchesInRange(request) && !force) {
+                log.error("{} Some of those batches were already delivered, so restarting it would re-send rows that are already committed at the target. "
+                        + "This load will not progress on its own: either truncate the target table and force recovery, or cancel the load.",
+                        describeStuckRequest(request));
+                continue;
+            }
+            log.warn("{} That cannot happen on a completed extract, so it is being re-queued for extraction.", describeStuckRequest(request));
+            restartExtractRequest(request);
+            restarted++;
+        }
+        if (restarted > 0) {
+            log.warn("Recovered {} stuck extract request(s)", restarted);
+        }
+        return restarted;
+    }
+
+    protected String describeStuckRequest(ExtractRequest request) {
+        return String.format("Extract request %d for table %s (load %d, node %s) is marked %s with %d of %d rows extracted, while batches %d through %d are "
+                + "still requested.", request.getRequestId(), request.getTableName(), request.getLoadId(), request.getNodeId(), ExtractStatus.OK.name(),
+                request.getExtractedRows(), request.getRows(), request.getStartBatchId(), request.getEndBatchId());
+    }
+
+    /** Load the batch range and any child requests, then hand off to {@link #restartExtractRequest(List, ExtractRequest, List)}. */
+    protected void restartExtractRequest(ExtractRequest request) {
+        List<OutgoingBatch> batches = outgoingBatchService.getOutgoingBatchRange(request.getStartBatchId(), request.getEndBatchId()).getBatches();
+        List<ExtractRequest> childRequests = request.getParentRequestId() == 0 ? getExtractChildRequestsForNode(request) : null;
+        restartExtractRequest(batches, request, childRequests);
+    }
+
     protected void restartExtractRequest(List<OutgoingBatch> batches, ExtractRequest request, List<ExtractRequest> childRequests) {
         /*
          * This extract request was interrupted and must start over

@@ -89,10 +89,34 @@ public DataExtractorServiceSqlMap(IDatabasePlatform platform,
 
         putSql("updateOutgoingBatchesForSetupThreadSql", "update $(outgoing_batch) set thread_id = ? where load_id = ? and summary like ? and thread_id is null");
 
+        // extracted_rows / extracted_millis are reset alongside the transfer and load statistics. Leaving them behind makes a
+        // restarted request report extraction counters from the run that was interrupted, which is the misleading state this
+        // recovery path exists to clear.
         putSql("restartExtractRequest", "update $(extract_request) set last_transferred_batch_id = null, transferred_rows = 0, transferred_millis = 0, "
-                + "last_loaded_batch_id = null, loaded_rows = 0, loaded_millis = 0, parent_request_id = 0, status = ? "
+                + "last_loaded_batch_id = null, loaded_rows = 0, loaded_millis = 0, extracted_rows = 0, extracted_millis = 0, "
+                + "parent_request_id = 0, status = ? "
                 + "where request_id = ? and node_id = ?");
 
+        putSql("updateExtractRequestExtractedStats", "update $(extract_request) set extracted_rows = extracted_rows + ?, "
+                + "extracted_millis = extracted_millis + ?, last_update_time = ? where request_id = ?");
+
+        putSql("selectRequestedBatchesForExtractRequestSql",
+                "select batch_id from $(outgoing_batch) where node_id = ? and batch_id between ? and ? and status = 'RQ'");
+
+        putSql("selectDeliveredBatchesForExtractRequestSql",
+                "select batch_id from $(outgoing_batch) where node_id = ? and batch_id between ? and ? and status in ('OK','IG')");
+
+        /*
+         * A request marked OK whose range still contains RQ batches cannot have completed: MultiBatchStagingWriter.close() advances every remaining batch, so a
+         * finished extract leaves none at RQ. loaded_time is null excludes requests that legitimately finished and were loaded, and the last_update_time bound
+         * avoids racing a status write that is still in flight.
+         */
+        putSql("selectStuckExtractRequestsSql", "select * from $(extract_request) r where r.source_node_id = ? and r.status = ? "
+                + "and r.loaded_time is null and r.parent_request_id = 0 and r.last_update_time < ? "
+                + "and exists (select 1 from $(outgoing_batch) b where b.node_id = r.node_id "
+                + "and b.batch_id between r.start_batch_id and r.end_batch_id and b.status = 'RQ') "
+                + "order by r.load_id asc, r.request_id asc");
+
         putSql("cancelExtractRequests", "update $(extract_request) set status=?, last_update_time=?, loaded_time=? where load_id = ? and source_node_id = ? and (status != ? or loaded_time is null)");
 
         putSql("countIncompleteExtractRequestsByLoadId", "select count(*) from $(extract_request) where load_id = ? and source_node_id = ? and parent_request_id = 0 and status != 'OK'");

@@ -74,6 +74,15 @@ protected boolean isApplicable(NodeCommunication nodeCommunication) {
     protected void updateExtractRequestsForThreading() {
     }
 
+    /**
+     * Not applicable to file sync, for the same reason as {@link #updateExtractRequestsForThreading()}: this service inherits {@code queueWork} and would
+     * otherwise scan and restart ordinary data extract requests, then resolve their staging through the file-sync {@code getStagedResource} override.
+     */
+    @Override
+    public int recoverStuckExtractRequests(boolean force) {
+        return 0;
+    }
+
     @Override
     protected boolean canProcessExtractRequest(ExtractRequest request, CommunicationType communicationType) {
         return request.getTableName().equalsIgnoreCase(TableConstants.getTableName(tablePrefix, TableConstants.SYM_FILE_SNAPSHOT));

@@ -1037,6 +1037,26 @@ initial.load.use.extract.job.enabled=true
 # Type: integer
 initial.load.extract.thread.per.server.count=20
 
+# Return an extract request to a requested state when it is marked complete but its batches are still requested.
+# That combination cannot occur on a completed extract, because closing the extract advances every batch in the
+# range; it is the state a load is left in when the JVM is interrupted mid-extract, and nothing else recovers it.
+# Left alone, the load makes no further progress and raises no error, so it is indistinguishable from a slow load.
+# A request whose range contains batches that were already delivered is reported but not restarted automatically,
+# because restarting re-sends rows that are already committed at the target.
+#
+# DatabaseOverridable: true
+# Tags: load
+# Type: boolean
+initial.load.extract.request.recovery.enabled=true
+
+# How long an extract request must have been untouched before recovery will consider it stuck.  This only exists to
+# avoid racing a status update that is still in flight, so it does not need to be long.
+#
+# DatabaseOverridable: true
+# Tags: load
+# Type: integer
+initial.load.extract.request.recovery.threshold.ms=300000
+
 # The number of milliseconds that the initial load extract job can run a thread
 # for processing extracting requests.
 #