Fixes open-metadata#27852: propagate tolerations from CronOMJob to scheduled OMJob (open-metadata#27955)

CronOMJobReconciler.deepCopyPodSpec was copying nodeSelector but
silently dropping tolerations when generating an OMJob from a
CronOMJob template. Manual runs worked because they go straight
through K8sPipelineClient.buildOMJob, but scheduled runs went
through this deep-copy and lost the field, leaving pods Pending
on tainted nodes.

Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com>

Author: pmbrull

openmetadata-k8s-operator/src/main/java/org/openmetadata/operator/controller/CronOMJobReconciler.java

@@ -352,6 +352,8 @@ private OMJobSpec.OMJobPodSpec deepCopyPodSpec(
     copy.setResources(source.getResources());
     copy.setNodeSelector(
         source.getNodeSelector() != null ? new HashMap<>(source.getNodeSelector()) : null);
+    copy.setTolerations(
+        source.getTolerations() != null ? new ArrayList<>(source.getTolerations()) : null);
     copy.setSecurityContext(source.getSecurityContext());
     copy.setLabels(source.getLabels() != null ? new HashMap<>(source.getLabels()) : null);
     copy.setAnnotations(

openmetadata-k8s-operator/src/test/java/org/openmetadata/operator/unit/CronOMJobReconcilerTest.java

@@ -17,23 +17,28 @@
 import static org.mockito.Mockito.*;
 
 import io.fabric8.kubernetes.api.model.ObjectMeta;
+import io.fabric8.kubernetes.api.model.Toleration;
+import io.fabric8.kubernetes.api.model.TolerationBuilder;
 import io.fabric8.kubernetes.client.KubernetesClient;
 import io.fabric8.kubernetes.client.dsl.MixedOperation;
 import io.fabric8.kubernetes.client.dsl.NamespaceableResource;
 import io.fabric8.kubernetes.client.dsl.Resource;
 import io.javaoperatorsdk.operator.api.reconciler.Context;
 import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
 import java.time.Instant;
+import java.util.List;
 import java.util.Map;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
 import org.openmetadata.operator.controller.CronOMJobReconciler;
 import org.openmetadata.operator.model.CronOMJobResource;
 import org.openmetadata.operator.model.CronOMJobSpec;
 import org.openmetadata.operator.model.CronOMJobStatus;
+import org.openmetadata.operator.model.OMJobResource;
 import org.openmetadata.operator.model.OMJobSpec;
 
 @ExtendWith(MockitoExtension.class)
@@ -158,6 +163,79 @@ void testReconcileWithNullSpec() {
     assertEquals("Missing schedule", cronOMJob.getStatus().getMessage());
   }
 
+  @Test
+  @SuppressWarnings("unchecked")
+  void testReconcilePropagatesTolerationsFromCronOMJobToScheduledOMJob() {
+    List<Toleration> tolerations =
+        List.of(
+            new TolerationBuilder()
+                .withKey("dedicated")
+                .withOperator("Equal")
+                .withValue("ingestion")
+                .withEffect("NoSchedule")
+                .build(),
+            new TolerationBuilder()
+                .withKey("gpu")
+                .withOperator("Exists")
+                .withEffect("NoExecute")
+                .withTolerationSeconds(300L)
+                .build());
+
+    cronOMJob.getSpec().getOmJobSpec().getMainPodSpec().setTolerations(tolerations);
+    cronOMJob.getSpec().getOmJobSpec().getExitHandlerSpec().setTolerations(tolerations);
+
+    cronOMJob.getSpec().setSchedule("* * * * *");
+    cronOMJob.getSpec().setStartingDeadlineSeconds(86400);
+
+    reconciler.reconcile(cronOMJob, context);
+
+    ArgumentCaptor<OMJobResource> captor = ArgumentCaptor.forClass(OMJobResource.class);
+    verify(mixedOp).resource(captor.capture());
+    OMJobResource scheduled = captor.getValue();
+
+    assertNotNull(scheduled.getSpec(), "Scheduled OMJob spec must not be null");
+    assertNotNull(
+        scheduled.getSpec().getMainPodSpec().getTolerations(),
+        "Main pod tolerations must be propagated from CronOMJob template");
+    assertEquals(2, scheduled.getSpec().getMainPodSpec().getTolerations().size());
+    assertEquals(
+        "dedicated", scheduled.getSpec().getMainPodSpec().getTolerations().get(0).getKey());
+    assertEquals(
+        "NoSchedule", scheduled.getSpec().getMainPodSpec().getTolerations().get(0).getEffect());
+    assertEquals("gpu", scheduled.getSpec().getMainPodSpec().getTolerations().get(1).getKey());
+    assertEquals(
+        Long.valueOf(300L),
+        scheduled.getSpec().getMainPodSpec().getTolerations().get(1).getTolerationSeconds());
+
+    assertNotNull(
+        scheduled.getSpec().getExitHandlerSpec().getTolerations(),
+        "Exit handler tolerations must be propagated from CronOMJob template");
+    assertEquals(2, scheduled.getSpec().getExitHandlerSpec().getTolerations().size());
+  }
+
+  @Test
+  @SuppressWarnings("unchecked")
+  void testReconcileLeavesTolerationsNullWhenSourceHasNone() {
+    cronOMJob.getSpec().getOmJobSpec().getMainPodSpec().setTolerations(null);
+    cronOMJob.getSpec().getOmJobSpec().getExitHandlerSpec().setTolerations(null);
+
+    cronOMJob.getSpec().setSchedule("* * * * *");
+    cronOMJob.getSpec().setStartingDeadlineSeconds(86400);
+
+    reconciler.reconcile(cronOMJob, context);
+
+    ArgumentCaptor<OMJobResource> captor = ArgumentCaptor.forClass(OMJobResource.class);
+    verify(mixedOp).resource(captor.capture());
+    OMJobResource scheduled = captor.getValue();
+
+    assertNull(
+        scheduled.getSpec().getMainPodSpec().getTolerations(),
+        "Main pod tolerations must remain null when CronOMJob has none");
+    assertNull(
+        scheduled.getSpec().getExitHandlerSpec().getTolerations(),
+        "Exit handler tolerations must remain null when CronOMJob has none");
+  }
+
   private CronOMJobResource createTestCronOMJob(String name, String schedule) {
     CronOMJobResource cronOMJob = new CronOMJobResource();