code: private void handleLockHeld(String taskId, ShardedJobStateImpl<T> jobState,
        IncrementalTaskState<T> taskState) {
      long currentTime = System.currentTimeMillis();
      int sliceTimeoutMillis = jobState.getSettings().getSliceTimeoutMillis();
      long lockExpiration = taskState.getLockInfo().lockedSince() + sliceTimeoutMillis;
      boolean wasRequestCompleted = wasRequestCompleted(taskState.getLockInfo().getRequestId());
  
      if (lockExpiration > currentTime && !wasRequestCompleted) {
        // if lock was not expired AND not abandon reschedule in 1 minute.
        long eta = Math.min(lockExpiration, currentTime + 60_000);
        scheduleWorkerTask(null, jobState.getSettings(), taskState, eta);
        log.info("Lock for " + taskId + " is being held. Will retry after " + (eta - currentTime));
      } else {
        ShardRetryState<T> retryState;
        if (wasRequestCompleted) {
          retryState = handleSliceFailure(jobState, taskState, new RuntimeException(
              "Resuming after abandon lock for " + taskId + " on slice: "
                  + taskState.getSequenceNumber()), true);
        } else {
          retryState = handleShardFailure(jobState, taskState, new RuntimeException(
            "Lock for " + taskId + " expired on slice: " + taskState.getSequenceNumber()));
        }
        updateTask(jobState, taskState, retryState, false);
      }
    }
docs: Handle a locked slice case.