feat(history): Migrate C7 external task logs to C8 job history records

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

@@ -7,12 +7,11 @@
  */
 package io.camunda.migration.data;
 
-import static io.camunda.migration.data.MigratorMode.LIST_SKIPPED;
-import static io.camunda.migration.data.MigratorMode.MIGRATE;
 import static io.camunda.migration.data.impl.persistence.IdKeyMapper.TYPE;
 import static io.camunda.migration.data.impl.persistence.IdKeyMapper.getHistoryTypes;
 
 import io.camunda.migration.data.config.C8DataSourceConfigured;
+import io.camunda.migration.data.impl.history.migrator.HistoryEntityMigrator;
 import io.camunda.migration.data.impl.history.migrator.DecisionDefinitionMigrator;
 import io.camunda.migration.data.impl.history.migrator.DecisionInstanceMigrator;
 import io.camunda.migration.data.impl.history.migrator.DecisionRequirementsMigrator;
@@ -24,10 +23,17 @@
 import io.camunda.migration.data.impl.history.migrator.UserTaskMigrator;
 import io.camunda.migration.data.impl.history.migrator.VariableMigrator;
 import io.camunda.migration.data.impl.history.migrator.AuditLogMigrator;
+import io.camunda.migration.data.impl.history.migrator.ExternalTaskMigrator;
 import io.camunda.migration.data.impl.clients.DbClient;
+import io.camunda.migration.data.impl.history.EntitySkippedException;
+import io.camunda.migration.data.impl.logging.HistoryMigratorLogs;
+import io.camunda.migration.data.impl.persistence.IdKeyMapper;
 import io.camunda.migration.data.impl.util.ExceptionUtils;
 import io.camunda.migration.data.impl.util.PrintUtils;
+import java.util.ArrayList;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Set;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Conditional;
 import org.springframework.stereotype.Component;
@@ -36,8 +42,6 @@
 @Conditional(C8DataSourceConfigured.class)
 public class HistoryMigrator {
 
-  // Migrator Services
-
   @Autowired
   protected FormMigrator formMigrator;
 
@@ -72,112 +76,101 @@ public class HistoryMigrator {
   protected AuditLogMigrator auditLogMigrator;
 
   @Autowired
-  protected DbClient dbClient;
-
-  protected MigratorMode mode = MIGRATE;
+  protected ExternalTaskMigrator externalTaskMigrator;
 
-  protected List<TYPE> requestedEntityTypes;
+  @Autowired
+  protected DbClient dbClient;
 
-  public void start() {
+  protected List<HistoryEntityMigrator<?, ?>> getMigrators() {
+    return List.of(
+        formMigrator,
+        processDefinitionMigrator,
+        processInstanceMigrator,
+        flowNodeMigrator,
+        userTaskMigrator,
+        variableMigrator,
+        externalTaskMigrator,
+        incidentMigrator,
+        decisionRequirementsMigrator,
+        decisionDefinitionMigrator,
+        decisionInstanceMigrator,
+        auditLogMigrator
+    );
+  }
+
+  public void printSkippedHistoryEntities(List<IdKeyMapper.TYPE> requestedEntityTypes) {
     try {
       ExceptionUtils.setContext(ExceptionUtils.ExceptionContext.HISTORY);
-      if (LIST_SKIPPED.equals(mode)) {
-        printSkippedHistoryEntities();
+
+      if (requestedEntityTypes == null || requestedEntityTypes.isEmpty()) {
+        getHistoryTypes().forEach(this::printSkippedEntitiesForType);
       } else {
-        migrate();
+        requestedEntityTypes.forEach(this::printSkippedEntitiesForType);
       }
+
     } finally {
       ExceptionUtils.clearContext();
     }
   }
 
-  protected void printSkippedHistoryEntities() {
-    if(requestedEntityTypes == null ||  requestedEntityTypes.isEmpty()) {
-      getHistoryTypes().forEach(this::printSkippedEntitiesForType);
-    } else {
-      requestedEntityTypes.forEach(this::printSkippedEntitiesForType);
-    }
-  }
-
   protected void printSkippedEntitiesForType(TYPE type) {
     PrintUtils.printSkippedInstancesHeader(dbClient.countSkippedByType(type), type);
     dbClient.listSkippedEntitiesByType(type);
   }
 
-  public void migrate() {
-    migrateForms();
-    migrateProcessDefinitions();
-    migrateProcessInstances();
-    migrateFlowNodes();
-    migrateUserTasks();
-    migrateVariables();
-    migrateIncidents();
-    migrateDecisionRequirementsDefinitions();
-    migrateDecisionDefinitions();
-    migrateDecisionInstances();
-    migrateAuditLogs();
-  }
+  public void retry() {
+    try {
+      ExceptionUtils.setContext(ExceptionUtils.ExceptionContext.HISTORY);
 
-  public void migrateForms() {
-    formMigrator.migrateAll();
-  }
+      getMigrators().forEach(migrator -> migrator.retry()
+          .forEach(HistoryMigratorLogs::logSkippingWarn));
 
-  public void migrateProcessDefinitions() {
-    processDefinitionMigrator.migrateAll();
+    } finally {
+      ExceptionUtils.clearContext();
+    }
   }
 
-  public void migrateProcessInstances() {
-    processInstanceMigrator.migrateAll();
-  }
+  public void migrate() {
+    try {
+      ExceptionUtils.setContext(ExceptionUtils.ExceptionContext.HISTORY);
 
-  public void migrateFlowNodes() {
-    flowNodeMigrator.migrateAll();
-  }
+      getMigrators().forEach(HistoryEntityMigrator::migrate);
 
-  public void migrateUserTasks() {
-    userTaskMigrator.migrateAll();
-  }
+      List<EntitySkippedException> permanentlySkippedExceptions = new ArrayList<>();
 
-  public void migrateVariables() {
-    variableMigrator.migrateAll();
-  }
+      long previousSkippedCount = Long.MAX_VALUE;
+      long currentSkippedCount = dbClient.countSkipped();
 
-  public void migrateIncidents() {
-    incidentMigrator.migrateAll();
-  }
+      while (currentSkippedCount > 0 && currentSkippedCount < previousSkippedCount) {
+        permanentlySkippedExceptions.clear(); // Only keep exceptions from the latest retry loop
+        previousSkippedCount = currentSkippedCount;
+        getMigrators().forEach(migrator -> permanentlySkippedExceptions.addAll(migrator.retry()));
+        currentSkippedCount = dbClient.countSkipped();
+      }
 
-  public void migrateDecisionRequirementsDefinitions() {
-    decisionRequirementsMigrator.migrateAll();
+      // Log all permanently skipped entities after retrying
+      permanentlySkippedExceptions.forEach(HistoryMigratorLogs::logSkippingWarn);
+    } finally {
+      ExceptionUtils.clearContext();
+    }
   }
 
-  public void migrateDecisionDefinitions() {
-    decisionDefinitionMigrator.migrateAll();
-  }
+  public void migrateByType(TYPE type) {
+    HistoryEntityMigrator<?, ?> migrator = getMigrators().stream()
+        .filter(m -> m.getType() == type)
+        .findFirst()
+        .orElseThrow(() -> new IllegalArgumentException("No migrator found for type: " + type));
 
-  public void migrateDecisionInstances() {
-    decisionInstanceMigrator.migrateAll();
+    migrator.migrate();
   }
 
-  public void migrateAuditLogs() {
-    auditLogMigrator.migrateAll();
-  }
+  public void retryByType(TYPE type) {
+    HistoryEntityMigrator<?, ?> migrator = getMigrators().stream()
+        .filter(m -> m.getType() == type)
+        .findFirst()
+        .orElseThrow(() -> new IllegalArgumentException("No migrator found for type: " + type));
 
-  public void setRequestedEntityTypes(List<TYPE> requestedEntityTypes) {
-    this.requestedEntityTypes = requestedEntityTypes;
+    migrator.retry();
   }
 
-  public void setMode(MigratorMode mode) {
-    this.mode = mode;
-    formMigrator.setMode(mode);
-    decisionDefinitionMigrator.setMode(mode);
-    decisionInstanceMigrator.setMode(mode);
-    decisionRequirementsMigrator.setMode(mode);
-    flowNodeMigrator.setMode(mode);
-    incidentMigrator.setMode(mode);
-    processDefinitionMigrator.setMode(mode);
-    processInstanceMigrator.setMode(mode);
-    userTaskMigrator.setMode(mode);
-    variableMigrator.setMode(mode);
-    auditLogMigrator.setMode(mode);
-  }
 }