fix(transfer): Prevent JobPollingService thread crash on stale job claims (#1491)

This PR fixes a critical process-terminating thread crash in the
JobPollingService loop that occurs when a transfer worker attempts to
claim a job that has already been claimed and has credentials stored by
another worker instance.

**Problem**
In a distributed, highly concurrent deployment with multiple worker
instances, a worker can poll a job ID from an eventually consistent
database index that looks free (CREDS_AVAILABLE) but has actually
already been claimed by a faster peer instance.

When our worker performs a strongly consistent read (store.findJob) to
retrieve the job details, it detects the instanceId mismatch (indicating
the job belongs to another worker) and marks the job's state as CANCELED
in memory.

However, in the old implementation:

1. JobPollingService.tryToClaimJob failed to verify whether the
retrieved existingJob was null or canceled, proceeding blindly to build
the updatedJob to claim it.
2. Constructing this job object to transition to state
CREDS_ENCRYPTION_KEY_GENERATED threw a validation exception
(IllegalStateException) because credentials (encryptedAuthData) had
already been set by the peer worker.
3. Because the PortabilityJob builder execution occurred outside the
main try-catch block in tryToClaimJob, this exception was uncaught.
4. This uncaught thread failure propagated up, terminating the periodic
JobPollingService thread and causing the entire container sandbox to
crash.

**Solution**
Implemented two layers of safety in JobPollingService.java to resolve
this:

1. Proactive Prevention (Early Abort): Added null and CANCELED state
checks immediately after retrieving the job from the JobStore. If the
job has been deleted or marked canceled (which happens on instanceId
mismatch), the worker aborts the claim attempt early and returns false
safely.
2. Reactive Safety (Validation Safety Net): Wrapped the PortabilityJob
builder execution in a local try-catch block to safely handle any
unexpected IllegalStateException validation errors during object
construction, returning false (handled failure) instead of propagating
and crashing the thread.

These changes ensure that failing to claim a job (due to losing a race)
is treated as a handled, temporary failure, allowing the worker to
complete the current polling iteration normally and try again in the
next cycle instead of crashing.

Author: yukthagaduputi

portability-transfer/src/main/java/org/datatransferproject/transfer/JobPollingService.java

@@ -152,6 +152,19 @@ private void pollForUnassignedJob() {
   private boolean tryToClaimJob(UUID jobId, WorkerKeyPair keyPair) {
     // Lookup the job so we can append to its existing properties.
     PortabilityJob existingJob = store.findJob(jobId);
+    if (existingJob == null) {
+      monitor.debug(() -> format("JobPollingService: tryToClaimJob: jobId: %s not found", jobId));
+      return false;
+    }
+    if (existingJob.state() == PortabilityJob.State.CANCELED) {
+      monitor.debug(
+          () ->
+              format(
+                  "JobPollingService: tryToClaimJob: jobId: %s is canceled (likely instance"
+                      + " mismatch)",
+                  jobId));
+      return false;
+    }
     monitor.debug(() -> format("JobPollingService: tryToClaimJob: jobId: %s", existingJob));
 
     // TODO: Consider moving this check earlier in the flow
@@ -166,18 +179,29 @@ private boolean tryToClaimJob(UUID jobId, WorkerKeyPair keyPair) {
     }
     String serializedKey = publicKeySerializer.serialize(keyPair.getEncodedPublicKey());
 
-    PortabilityJob updatedJob =
-        existingJob
-            .toBuilder()
-            .setAndValidateJobAuthorization(
-                existingJob
-                    .jobAuthorization()
-                    .toBuilder()
-                    .setInstanceId(keyPair.getInstanceId())
-                    .setAuthPublicKey(serializedKey)
-                    .setState(JobAuthorization.State.CREDS_ENCRYPTION_KEY_GENERATED)
-                    .build())
-            .build();
+    PortabilityJob updatedJob;
+    try {
+      updatedJob =
+          existingJob.toBuilder()
+              .setAndValidateJobAuthorization(
+                  existingJob
+                      .jobAuthorization()
+                      .toBuilder()
+                      .setInstanceId(keyPair.getInstanceId())
+                      .setAuthPublicKey(serializedKey)
+                      .setState(JobAuthorization.State.CREDS_ENCRYPTION_KEY_GENERATED)
+                      .build())
+              .build();
+    } catch (IllegalStateException e) {
+      monitor.debug(
+          () ->
+              format(
+                  "Could not build updated job object for %s. Validation failed. Error msg: %s",
+                  jobId, e.getMessage()),
+          e);
+      return false;
+    }
+
     // Attempt to 'claim' this job by validating it is still in state CREDS_AVAILABLE as we
     // update it to state CREDS_ENCRYPTION_KEY_GENERATED, along with our key. If another transfer
     // instance polled the same job, and already claimed it, it will have updated the job's state