feat(history): migrate async-before/after job logs from C7 to C8

related to #5331

- Add HISTORY_JOB type to IdKeyMapper.TYPE with HistoricJobLogEventEntity class mapping
- Create JobMigrator: migrates C7 HistoricJobLog entries tracked by job ID (deduplicates multiple log entries per job)
- Create JobTransformer: EntityInterceptor mapping C7 job log fields to C8 JobDbModel (state, retries, error, timing, etc.)
- Add fetchAndHandleHistoricJobLogs() and getHistoricJobLog() to C7Client
- Add insertJob() to C8Client
- Add of(HistoricJobLog) factory method to C7Entity using getJobId() as tracking key
- Add HistoricJobLog constructor to EntitySkippedException
- Update IncidentMigrator: resolve jobKey for failedJob incidents from HISTORY_JOB tracking table; skip incident if the referenced job was explicitly skipped
- Update HistoryMigrator: call migrateJobs() before migrateIncidents()
- Enable previously-disabled test shouldNotMigrateIncidentsWhenJobIsSkipped (changed HISTORY_FLOW_NODE to HISTORY_JOB)
- Add HistoryJobTest integration tests covering deduplication and incident FK population
- Add asyncBeforeUserTaskProcess.bpmn for testing successfully-completed async-before jobs

Co-authored-by: yanavasileva <26868499+yanavasileva@users.noreply.github.com>

Author: Copilot

data-migrator/core/src/main/java/io/camunda/migration/data/HistoryMigrator.java

@@ -19,6 +19,7 @@
 import io.camunda.migration.data.impl.history.migrator.FlowNodeMigrator;
 import io.camunda.migration.data.impl.history.migrator.FormMigrator;
 import io.camunda.migration.data.impl.history.migrator.IncidentMigrator;
+import io.camunda.migration.data.impl.history.migrator.JobMigrator;
 import io.camunda.migration.data.impl.history.migrator.ProcessDefinitionMigrator;
 import io.camunda.migration.data.impl.history.migrator.ProcessInstanceMigrator;
 import io.camunda.migration.data.impl.history.migrator.UserTaskMigrator;
@@ -59,6 +60,9 @@ public class HistoryMigrator {
   @Autowired
   protected IncidentMigrator incidentMigrator;
 
+  @Autowired
+  protected JobMigrator jobMigrator;
+
   @Autowired
   protected DecisionRequirementsMigrator decisionRequirementsMigrator;
 
@@ -111,6 +115,7 @@ public void migrate() {
     migrateFlowNodes();
     migrateUserTasks();
     migrateVariables();
+    migrateJobs();
     migrateIncidents();
     migrateDecisionRequirementsDefinitions();
     migrateDecisionDefinitions();
@@ -146,6 +151,10 @@ public void migrateIncidents() {
     incidentMigrator.migrateAll();
   }
 
+  public void migrateJobs() {
+    jobMigrator.migrateAll();
+  }
+
   public void migrateDecisionRequirementsDefinitions() {
     decisionRequirementsMigrator.migrateAll();
   }
@@ -174,6 +183,7 @@ public void setMode(MigratorMode mode) {
     decisionRequirementsMigrator.setMode(mode);
     flowNodeMigrator.setMode(mode);
     incidentMigrator.setMode(mode);
+    jobMigrator.setMode(mode);
     processDefinitionMigrator.setMode(mode);
     processInstanceMigrator.setMode(mode);
     userTaskMigrator.setMode(mode);

data-migrator/core/src/main/java/io/camunda/migration/data/impl/clients/C7Client.java

@@ -42,6 +42,7 @@
 import org.camunda.bpm.engine.history.HistoricActivityInstance;
 import org.camunda.bpm.engine.history.HistoricDecisionInstance;
 import org.camunda.bpm.engine.history.HistoricIncident;
+import org.camunda.bpm.engine.history.HistoricJobLog;
 import org.camunda.bpm.engine.history.HistoricProcessInstance;
 import org.camunda.bpm.engine.history.HistoricTaskInstance;
 import org.camunda.bpm.engine.history.HistoricVariableInstance;
@@ -52,6 +53,7 @@
 import org.camunda.bpm.engine.impl.HistoricActivityInstanceQueryImpl;
 import org.camunda.bpm.engine.impl.HistoricDecisionInstanceQueryImpl;
 import org.camunda.bpm.engine.impl.HistoricIncidentQueryImpl;
+import org.camunda.bpm.engine.impl.HistoricJobLogQueryImpl;
 import org.camunda.bpm.engine.impl.HistoricProcessInstanceQueryImpl;
 import org.camunda.bpm.engine.impl.HistoricTaskInstanceQueryImpl;
 import org.camunda.bpm.engine.impl.HistoricVariableInstanceQueryImpl;
@@ -571,6 +573,39 @@ public UserOperationLogEntry getUserOperationLogEntry(String c7Id) {
     return callApi(query::singleResult, format(FAILED_TO_FETCH_HISTORIC_ELEMENT, "UserOperationLogEntry", c7Id));
   }
 
+  /**
+   * Fetches the first historic job log entry for the given job ID, ordered by timestamp ascending.
+   * Used for retry mode, where the job ID is the tracking key.
+   */
+  public HistoricJobLog getHistoricJobLog(String jobId) {
+    HistoricJobLogQueryImpl query = (HistoricJobLogQueryImpl) historyService.createHistoricJobLogQuery()
+        .jobId(jobId)
+        .orderByTimestamp()
+        .asc()
+        .orderByJobId()
+        .asc();
+    List<HistoricJobLog> results = callApi(query::list,
+        format(FAILED_TO_FETCH_HISTORIC_ELEMENT, "HistoricJobLog with jobId", jobId));
+    return results.isEmpty() ? null : results.getFirst();
+  }
+
+  /**
+   * Processes historic job log entries with pagination using the provided callback consumer.
+   * The {@code ignoredCreatedAfter} parameter is not supported by the query and is ignored;
+   * deduplication is handled via the tracking table.
+   */
+  public void fetchAndHandleHistoricJobLogs(Consumer<HistoricJobLog> callback, Date ignoredCreatedAfter) {
+    HistoricJobLogQueryImpl query = (HistoricJobLogQueryImpl) historyService.createHistoricJobLogQuery()
+        .orderByTimestamp()
+        .asc()
+        .orderByJobId()
+        .asc();
+
+    new Pagination<HistoricJobLog>().pageSize(properties.getPageSize())
+        .query(query)
+        .callback(callback);
+  }
+
   /**
    * Processes tenant entities with pagination using the provided callback consumer.
    */

data-migrator/core/src/main/java/io/camunda/migration/data/impl/clients/C8Client.java

@@ -22,6 +22,7 @@
 import static io.camunda.migration.data.impl.logging.C8ClientLogs.FAILED_TO_INSERT_DECISION_REQUIREMENTS;
 import static io.camunda.migration.data.impl.logging.C8ClientLogs.FAILED_TO_INSERT_FLOW_NODE_INSTANCE;
 import static io.camunda.migration.data.impl.logging.C8ClientLogs.FAILED_TO_INSERT_INCIDENT;
+import static io.camunda.migration.data.impl.logging.C8ClientLogs.FAILED_TO_INSERT_JOB;
 import static io.camunda.migration.data.impl.logging.C8ClientLogs.FAILED_TO_INSERT_PROCESS_DEFINITION;
 import static io.camunda.migration.data.impl.logging.C8ClientLogs.FAILED_TO_INSERT_PROCESS_INSTANCE;
 import static io.camunda.migration.data.impl.logging.C8ClientLogs.FAILED_TO_INSERT_USER_TASK;
@@ -66,6 +67,7 @@
 import io.camunda.db.rdbms.sql.FlowNodeInstanceMapper;
 import io.camunda.db.rdbms.sql.FormMapper;
 import io.camunda.db.rdbms.sql.IncidentMapper;
+import io.camunda.db.rdbms.sql.JobMapper;
 import io.camunda.db.rdbms.sql.ProcessDefinitionMapper;
 import io.camunda.db.rdbms.sql.ProcessInstanceMapper;
 import io.camunda.db.rdbms.sql.UserTaskMapper;
@@ -77,6 +79,7 @@
 import io.camunda.db.rdbms.write.domain.FlowNodeInstanceDbModel;
 import io.camunda.db.rdbms.write.domain.FormDbModel;
 import io.camunda.db.rdbms.write.domain.IncidentDbModel;
+import io.camunda.db.rdbms.write.domain.JobDbModel;
 import io.camunda.db.rdbms.write.domain.ProcessDefinitionDbModel;
 import io.camunda.db.rdbms.write.domain.ProcessInstanceDbModel;
 import io.camunda.db.rdbms.write.domain.UserTaskDbModel;
@@ -146,6 +149,9 @@ public class C8Client {
   @Autowired(required = false)
   protected FormMapper formMapper;
 
+  @Autowired(required = false)
+  protected JobMapper jobMapper;
+
   /**
    * Creates a new process instance with the given BPMN process ID and variables.
    */
@@ -400,4 +406,11 @@ public void insertForm(FormDbModel dbModel) {
     callApi(() -> formMapper.insert(dbModel), "Failed to insert form");
   }
 
+  /**
+   * Inserts a Job into the database.
+   */
+  public void insertJob(JobDbModel dbModel) {
+    callApi(() -> jobMapper.insert(new BatchInsertDto(List.of(dbModel))), FAILED_TO_INSERT_JOB);
+  }
+
 }

data-migrator/core/src/main/java/io/camunda/migration/data/impl/history/C7Entity.java

@@ -13,6 +13,7 @@
 import org.camunda.bpm.engine.history.HistoricActivityInstance;
 import org.camunda.bpm.engine.history.HistoricDecisionInstance;
 import org.camunda.bpm.engine.history.HistoricIncident;
+import org.camunda.bpm.engine.history.HistoricJobLog;
 import org.camunda.bpm.engine.history.HistoricProcessInstance;
 import org.camunda.bpm.engine.history.HistoricTaskInstance;
 import org.camunda.bpm.engine.history.HistoricVariableInstance;
@@ -42,6 +43,7 @@ public static C7Entity<?> of(Object c7) {
       case HistoricDecisionInstance c7DecisionInstance -> of(c7DecisionInstance);
       case HistoricActivityInstance c7ActivityInstance -> of(c7ActivityInstance);
       case HistoricIncident c7Incident -> of(c7Incident);
+      case HistoricJobLog c7JobLog -> of(c7JobLog);
       case HistoricProcessInstance c7ProcessInstance -> of(c7ProcessInstance);
       case HistoricTaskInstance c7TaskInstance -> of(c7TaskInstance);
       case HistoricVariableInstance c7VariableInstance -> of(c7VariableInstance);
@@ -104,6 +106,10 @@ public static C7Entity<HistoricVariableInstance> of(HistoricVariableInstance c7E
     return new C7Entity<>(c7Entity.getId(), c7Entity.getCreateTime(), c7Entity);
   }
 
+  public static C7Entity<HistoricJobLog> of(HistoricJobLog c7Entity) {
+    return new C7Entity<>(c7Entity.getJobId(), c7Entity.getTimestamp(), c7Entity);
+  }
+
   public String getId() {
     return id;
   }

data-migrator/core/src/main/java/io/camunda/migration/data/impl/history/EntitySkippedException.java

@@ -12,6 +12,7 @@
 import org.camunda.bpm.engine.history.HistoricActivityInstance;
 import org.camunda.bpm.engine.history.HistoricDecisionInstance;
 import org.camunda.bpm.engine.history.HistoricIncident;
+import org.camunda.bpm.engine.history.HistoricJobLog;
 import org.camunda.bpm.engine.history.HistoricProcessInstance;
 import org.camunda.bpm.engine.history.HistoricTaskInstance;
 import org.camunda.bpm.engine.history.HistoricVariableInstance;
@@ -48,6 +49,10 @@ public EntitySkippedException(HistoricIncident c7Incident, String message) {
     this(C7Entity.of(c7Incident), message);
   }
 
+  public EntitySkippedException(HistoricJobLog c7JobLog, String message) {
+    this(C7Entity.of(c7JobLog), message);
+  }
+
   public EntitySkippedException(HistoricProcessInstance c7ProcessInstance, String message) {
     this(C7Entity.of(c7ProcessInstance), message);
   }

data-migrator/core/src/main/java/io/camunda/migration/data/impl/history/migrator/IncidentMigrator.java

@@ -8,15 +8,16 @@
 package io.camunda.migration.data.impl.history.migrator;
 
 import static io.camunda.migration.data.impl.logging.HistoryMigratorLogs.SKIP_REASON_MISSING_FLOW_NODE;
+import static io.camunda.migration.data.impl.logging.HistoryMigratorLogs.SKIP_REASON_MISSING_JOB_REFERENCE;
 import static io.camunda.migration.data.impl.logging.HistoryMigratorLogs.SKIP_REASON_MISSING_PROCESS_DEFINITION;
 import static io.camunda.migration.data.impl.logging.HistoryMigratorLogs.SKIP_REASON_MISSING_PROCESS_INSTANCE;
 import static io.camunda.migration.data.impl.logging.HistoryMigratorLogs.SKIP_REASON_MISSING_ROOT_PROCESS_INSTANCE;
 import static io.camunda.migration.data.impl.persistence.IdKeyMapper.TYPE.HISTORY_INCIDENT;
+import static io.camunda.migration.data.impl.persistence.IdKeyMapper.TYPE.HISTORY_JOB;
 import static io.camunda.migration.data.impl.persistence.IdKeyMapper.TYPE.HISTORY_PROCESS_INSTANCE;
 
 import io.camunda.db.rdbms.write.domain.IncidentDbModel;
 import io.camunda.db.rdbms.write.domain.IncidentDbModel.Builder;
-import io.camunda.migration.data.exception.EntityInterceptorException;
 import io.camunda.migration.data.impl.history.C7Entity;
 import io.camunda.migration.data.impl.history.EntitySkippedException;
 import io.camunda.migration.data.impl.logging.HistoryMigratorLogs;
@@ -76,8 +77,6 @@ public Long migrateTransactionally(HistoricIncident c7Incident) {
         if (processInstanceKey != null) {
           var flowNodeInstanceKey = findFlowNodeInstanceKey(c7Incident.getActivityId(), c7Incident.getProcessInstanceId());
           builder.flowNodeInstanceKey(flowNodeInstanceKey);
-//          .jobKey(jobDefinitionKey) // TODO when jobs are migrated
-
 
           String c7RootProcessInstanceId = c7Incident.getRootProcessInstanceId();
           if (c7RootProcessInstanceId != null && isMigrated(c7RootProcessInstanceId, HISTORY_PROCESS_INSTANCE)) {
@@ -89,29 +88,28 @@ public Long migrateTransactionally(HistoricIncident c7Incident) {
         }
       }
 
+      resolveJobKey(c7Incident, builder);
+
       IncidentDbModel dbModel = convert(C7Entity.of(c7Incident), builder);
 
-    if (dbModel.processDefinitionKey() == null) {
-      throw new EntitySkippedException(c7Incident, SKIP_REASON_MISSING_PROCESS_DEFINITION);
-    }
+      if (dbModel.processDefinitionKey() == null) {
+        throw new EntitySkippedException(c7Incident, SKIP_REASON_MISSING_PROCESS_DEFINITION);
+      }
 
-    if (dbModel.processInstanceKey() == null) {
+      if (dbModel.processInstanceKey() == null) {
         throw new EntitySkippedException(c7Incident, SKIP_REASON_MISSING_PROCESS_INSTANCE);
-    }
+      }
 
-    if (dbModel.rootProcessInstanceKey() == null) {
-      throw new EntitySkippedException(c7Incident, SKIP_REASON_MISSING_ROOT_PROCESS_INSTANCE);
-    }
+      if (dbModel.rootProcessInstanceKey() == null) {
+        throw new EntitySkippedException(c7Incident, SKIP_REASON_MISSING_ROOT_PROCESS_INSTANCE);
+      }
 
-    if (dbModel.flowNodeInstanceKey() == null) {
-      if (!c7Client.hasWaitingExecution(c7Incident.getProcessInstanceId(), c7Incident.getActivityId())) { // Activities on async before waiting state will not have a flow node instance key, but should not be skipped
-        throw new EntitySkippedException(c7Incident, SKIP_REASON_MISSING_FLOW_NODE);
+      if (dbModel.flowNodeInstanceKey() == null) {
+        if (!c7Client.hasWaitingExecution(c7Incident.getProcessInstanceId(), c7Incident.getActivityId())) {
+          throw new EntitySkippedException(c7Incident, SKIP_REASON_MISSING_FLOW_NODE);
+        }
       }
-    }
 
-    if (dbModel.jobKey() == null) {
-//      throw new EntitySkippedException(c7Incident, SKIP_REASON_MISSING_JOB_REFERENCE); // TODO when jobs are migrated
-    }
       c8Client.insertIncident(dbModel);
 
       return dbModel.incidentKey();
@@ -120,5 +118,51 @@ public Long migrateTransactionally(HistoricIncident c7Incident) {
     return null;
   }
 
+  /**
+   * Resolves and sets the job key on the incident builder for {@code failedJob} incidents.
+   * <p>
+   * If the incident is a {@code failedJob} incident and the associated C7 job has been tracked
+   * in the migration table:
+   * <ul>
+   *   <li>If the job was migrated (has a C8 key), sets the {@code jobKey} on the builder.</li>
+   *   <li>If the job was explicitly skipped (null C8 key), throws an
+   *       {@link EntitySkippedException} to skip this incident as well.</li>
+   * </ul>
+   * If the job is not yet tracked (job migration may not have run or the job type is not
+   * tracked), the incident proceeds without a job key.
+   * </p>
+   *
+   * @param c7Incident the Camunda 7 incident
+   * @param builder    the incident builder to set the job key on
+   * @throws EntitySkippedException if the associated job was explicitly skipped
+   */
+  protected void resolveJobKey(final HistoricIncident c7Incident, final Builder builder) {
+    if (!isFailedJobIncident(c7Incident)) {
+      return;
+    }
+    final String c7JobId = c7Incident.getConfiguration();
+    if (c7JobId == null) {
+      return;
+    }
+    if (dbClient.checkExistsByC7IdAndType(c7JobId, HISTORY_JOB)) {
+      final Long jobKey = dbClient.findC8KeyByC7IdAndType(c7JobId, HISTORY_JOB);
+      if (jobKey != null) {
+        builder.jobKey(jobKey);
+      } else {
+        throw new EntitySkippedException(c7Incident, SKIP_REASON_MISSING_JOB_REFERENCE);
+      }
+    }
+  }
+
+  /**
+   * Returns true if this incident was caused by a job failure.
+   *
+   * @param c7Incident the Camunda 7 incident
+   * @return true for {@code failedJob} incident type
+   */
+  protected boolean isFailedJobIncident(final HistoricIncident c7Incident) {
+    return "failedJob".equals(c7Incident.getIncidentType());
+  }
+
 }