Initial plan

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>

fix(history): update IncidentMigrator Javadoc to reflect job key validation

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

Author: Copilot

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

@@ -18,6 +18,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;
@@ -62,6 +63,9 @@ public class HistoryMigrator {
   @Autowired
   protected IncidentMigrator incidentMigrator;
 
+  @Autowired
+  protected JobMigrator jobMigrator;
+
   @Autowired
   protected DecisionRequirementsMigrator decisionRequirementsMigrator;
 
@@ -85,6 +89,7 @@ public class HistoryMigrator {
         flowNodeMigrator,
         userTaskMigrator,
         variableMigrator,
+        jobMigrator,
         incidentMigrator,
         decisionRequirementsMigrator,
         decisionDefinitionMigrator,

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;
@@ -56,6 +57,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;
@@ -570,6 +572,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

@@ -24,6 +24,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;
@@ -68,6 +69,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;
@@ -79,6 +81,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;
@@ -148,6 +151,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.
    */
@@ -444,4 +450,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

@@ -11,6 +11,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;
@@ -47,6 +48,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,10 +8,12 @@
 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.read.domain.FlowNodeInstanceDbQuery;
@@ -66,12 +68,10 @@ public IdKeyMapper.TYPE getType() {
    *   <li>Process instance key missing - skipped with {@code SKIP_REASON_MISSING_PROCESS_INSTANCE_KEY}</li>
    *   <li>Process definition not yet migrated - skipped with {@code SKIP_REASON_MISSING_PROCESS_DEFINITION}</li>
    *   <li>Root process instance not yet migrated (when part of a process hierarchy) - skipped with {@code SKIP_REASON_MISSING_ROOT_PROCESS_INSTANCE}</li>
+   *   <li>Referenced job explicitly skipped - skipped with {@code SKIP_REASON_MISSING_JOB_REFERENCE}</li>
    *   <li>Interceptor error during conversion - skipped with the exception message</li>
    * </ul>
    *
-   * <p><strong>Note:</strong> Flow node instance and job reference validations are currently disabled
-   * pending resolution of known issues. See code comments for details.
-   *
    * @param c7Incident the historic incident from Camunda 7 to be migrated
    * @throws EntityInterceptorException if an error occurs during entity conversion (handled internally, entity marked as skipped)
    */
@@ -104,6 +104,8 @@ public Long migrateTransactionally(HistoricIncident c7Incident) {
         }
       }
 
+      resolveJobKey(c7Incident, builder);
+
       IncidentDbModel dbModel = convert(C7Entity.of(c7Incident), builder);
 
     if (dbModel.processDefinitionKey() == null) {
@@ -154,5 +156,51 @@ protected Long findFlowNodeInstanceKey(String activityId, String processInstance
     }
   }
 
+  /**
+   * 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());
+  }
+
 }
 

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

@@ -0,0 +1,117 @@
+/*
+ * Copyright Camunda Services GmbH and/or licensed to Camunda Services GmbH under
+ * one or more contributor license agreements. See the NOTICE file distributed
+ * with this work for additional information regarding copyright ownership.
+ * Licensed under the Camunda License 1.0. You may not use this file
+ * except in compliance with the Camunda License 1.0.
+ */
+package io.camunda.migration.data.impl.history.migrator;
+
+import static io.camunda.migration.data.impl.logging.HistoryMigratorLogs.SKIP_REASON_MISSING_PROCESS_INSTANCE;
+import static io.camunda.migration.data.impl.logging.HistoryMigratorLogs.logMigratingJob;
+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 static io.camunda.migration.data.impl.util.ConverterUtil.getNextKey;
+
+import io.camunda.db.rdbms.write.domain.JobDbModel;
+import io.camunda.migration.data.impl.history.C7Entity;
+import io.camunda.migration.data.impl.history.EntitySkippedException;
+import io.camunda.migration.data.impl.logging.HistoryMigratorLogs;
+import io.camunda.search.entities.ProcessInstanceEntity;
+import org.camunda.bpm.engine.history.HistoricJobLog;
+import org.springframework.stereotype.Service;
+
+/**
+ * Service class responsible for migrating historic job log entries from Camunda 7 to Camunda 8.
+ * <p>
+ * Job logs in Camunda 7 record lifecycle events (creation, execution, failure, deletion) for each
+ * job. This migrator converts C7 job log entries to C8 {@link JobDbModel} records, tracking each
+ * C7 job by its job ID so that only one C8 record is created per C7 job.
+ * </p>
+ * <p>
+ * The C8 job key is stored in the migration tracking table keyed by the C7 job ID. This allows
+ * the {@link IncidentMigrator} to look up the C8 job key when migrating {@code failedJob}
+ * incidents.
+ * </p>
+ */
+@Service
+public class JobMigrator extends BaseMigrator<HistoricJobLog, JobDbModel> {
+
+  /**
+   * Migrates all historic job log entries from Camunda 7 to Camunda 8.
+   * <p>
+   * Processes log entries in ascending timestamp order. Since tracking is done by C7 job ID, only
+   * the first log entry encountered per job is migrated; subsequent entries for the same job are
+   * deduplicated via the tracking table.
+   * </p>
+   */
+  @Override
+  public void migrateAll() {
+    fetchMigrateOrRetry(
+        HISTORY_JOB,
+        c7Client::getHistoricJobLog,
+        c7Client::fetchAndHandleHistoricJobLogs
+    );
+  }
+
+  /**
+   * Migrates a single historic job log entry from Camunda 7 to Camunda 8.
+   * <p>
+   * Uses the C7 job ID as the tracking key, ensuring that only one C8 job record is created
+   * per C7 job across multiple log entries.
+   * </p>
+   *
+   * <p>Skip scenarios:
+   * <ul>
+   *   <li>Job already tracked in the migration table - silently skipped</li>
+   *   <li>Process instance not yet migrated - skipped with
+   *       {@code SKIP_REASON_MISSING_PROCESS_INSTANCE}</li>
+   * </ul>
+   *
+   * @param c7JobLog the Camunda 7 historic job log entry to migrate
+   * @return the C8 job key as a string, or {@code null} if already migrated
+   */
+  @Override
+  public Long migrateTransactionally(final HistoricJobLog c7JobLog) {
+    final String c7JobId = c7JobLog.getJobId();
+    if (shouldMigrate(c7JobId, HISTORY_JOB)) {
+      logMigratingJob(c7JobId);
+
+      final var jobKey = getNextKey();
+      final var builder = new JobDbModel.Builder().jobKey(jobKey);
+
+      final String c7ProcessInstanceId = c7JobLog.getProcessInstanceId();
+      final ProcessInstanceEntity processInstance = findProcessInstanceByC7Id(c7ProcessInstanceId);
+      if (processInstance != null) {
+        builder.processInstanceKey(processInstance.processInstanceKey());
+
+        final var processDefinitionKey = findProcessDefinitionKey(c7JobLog.getProcessDefinitionId());
+        builder.processDefinitionKey(processDefinitionKey);
+
+        final String c7RootProcessInstanceId = c7JobLog.getRootProcessInstanceId();
+        if (c7RootProcessInstanceId != null && isMigrated(c7RootProcessInstanceId, HISTORY_PROCESS_INSTANCE)) {
+          final ProcessInstanceEntity rootProcessInstance = findProcessInstanceByC7Id(c7RootProcessInstanceId);
+          if (rootProcessInstance != null && rootProcessInstance.processInstanceKey() != null) {
+            builder.rootProcessInstanceKey(rootProcessInstance.processInstanceKey());
+          }
+        }
+
+        final Long elementInstanceKey = findFlowNodeInstanceKey(
+            c7JobLog.getActivityId(), c7ProcessInstanceId);
+        builder.elementInstanceKey(elementInstanceKey);
+      }
+
+      final JobDbModel dbModel = convert(C7Entity.of(c7JobLog), builder);
+
+      if (dbModel.processInstanceKey() == null) {
+        throw new EntitySkippedException(c7JobLog, SKIP_REASON_MISSING_PROCESS_INSTANCE);
+      }
+
+      c8Client.insertJob(dbModel);
+
+      return jobKey;
+    }
+
+    return null;
+  }
+}