Merge pull request #14 from SAP/develop

Fix issues with migration resuming

commercedbsync/resources/commercedbsync-beans.xml

@@ -39,6 +39,7 @@
         <property name="durationinseconds" type="double"/>
         <property name="copyMethod" type="String"/>
         <property name="keyColumns" type="java.util.List&lt;String&gt;"/>
+        <property name="batchsize" type="int"/>
     </bean>
 
     <bean class="com.sap.cx.boosters.commercedbsync.service.DatabaseCopyBatch">

commercedbsync/resources/commercedbsync-items.xml

@@ -100,7 +100,18 @@
                     <persistence type="property"/>
                     <modifiers optional="false"/>
                     <defaultvalue>false</defaultvalue>
-                </attribute>            
+                </attribute>
+                <attribute qualifier="resumeMigration" type="boolean">
+                    <description>Resume a failed migration</description>
+                    <persistence type="property" />
+                    <modifiers optional="true"/>
+                    <defaultvalue>false</defaultvalue>
+                </attribute>
+                <attribute qualifier="migrationId" type="java.lang.String" >
+                    <description>Migration Id for the job</description>
+                    <persistence type="property" />
+                    <modifiers optional="true"/>
+                </attribute>
             </attributes>
         </itemtype>
 

commercedbsync/resources/commercedbsync-spring.xml

@@ -315,6 +315,7 @@
 
   <bean id="fullMigrationJob" class="com.sap.cx.boosters.commercedbsync.jobs.FullMigrationJob"
     parent="abstractMigrationJobPerformable">
+      <property name="modelService" ref="modelService"/>
   </bean>
 
 <!--  <bean id="defaultCMTRemoveInterceptor"-->

commercedbsync/resources/localization/commercedbsync-locales_en.properties

@@ -30,3 +30,9 @@ type.MigrationCronJob.maxWriterWorkers.description=Number of writer workers to b
 
 type.MigrationCronJob.batchSize.name=Batch Size
 type.MigrationCronJob.batchSize.description=Batch size used to query data
+
+type.FullMigrationCronJob.resumeMigration.name=Resume Migration
+type.FullMigrationCronJob.resumeMigration.description=
+
+type.FullMigrationCronJob.migrationId.name=Migration ID
+type.FullMigrationCronJob.migrationId.description=

commercedbsync/resources/sql/createSchedulerTablesHANA.sql

@@ -44,6 +44,7 @@ CREATE TABLE MIGRATIONTOOLKIT_TABLECOPYTASKS (
                                                  copymethod NVARCHAR(255) NULL,
                                                  keycolumns NVARCHAR(255) NULL,
                                                  durationinseconds numeric(10,2) NULL DEFAULT 0,
+                                                 batchsize int NOT NULL DEFAULT 1000,
                                                  PRIMARY KEY (migrationid, targetnodeid, pipelinename)
 );
 

commercedbsync/resources/sql/createSchedulerTablesMSSQL.sql

@@ -21,6 +21,7 @@ CREATE TABLE MIGRATIONTOOLKIT_TABLECOPYTASKS (
     copymethod NVARCHAR(255) NULL,
     keycolumns NVARCHAR(255) NULL,
     durationinseconds numeric(10,2) NULL DEFAULT 0,
+    batchsize int NOT NULL DEFAULT 1000,
     PRIMARY KEY (migrationid, targetnodeid, pipelinename)
 );
 

commercedbsync/resources/sql/createSchedulerTablesMYSQL.sql

@@ -21,6 +21,7 @@ CREATE TABLE MIGRATIONTOOLKIT_TABLECOPYTASKS
     copymethod             VARCHAR(255) NULL,
     keycolumns             VARCHAR(255) NULL,
     durationinseconds      numeric(10, 2) NULL DEFAULT 0,
+    batchsize              int          NOT NULL DEFAULT 1000,
     PRIMARY KEY (migrationid, targetnodeid, pipelinename)
 );
 #

commercedbsync/resources/sql/createSchedulerTablesORACLE.sql

@@ -27,6 +27,7 @@ CREATE TABLE MIGRATIONTOOLKIT_TABLECOPYTASKS (
     copymethod NVARCHAR2(255) NULL,
     keycolumns NVARCHAR2(255) NULL,
     durationinseconds number(10,2) DEFAULT 0 NULL,
+    batchsize number(10) DEFAULT 1000 NOT NULL,
     PRIMARY KEY (migrationid, targetnodeid, pipelinename)
 )
 /

commercedbsync/resources/sql/createSchedulerTablesPOSTGRESQL.sql

@@ -22,6 +22,7 @@ CREATE TABLE MIGRATIONTOOLKIT_TABLECOPYTASKS (
     copymethod VARCHAR(255) NULL,
     keycolumns VARCHAR(255) NULL,
     durationinseconds numeric(10,2) NULL DEFAULT 0,
+    batchsize int NOT NULL DEFAULT 1000,
     PRIMARY KEY (migrationid, targetnodeid, pipelinename)
 );
 

commercedbsync/src/com/sap/cx/boosters/commercedbsync/concurrent/impl/DefaultDataPipeFactory.java

@@ -84,7 +84,7 @@ public DataPipe<DataSet> create(CopyContext context, CopyContext.DataCopyItem it
                     try {
                         pipe.put(MaybeFinished.poison());
                     } catch (Exception p) {
-                        LOG.error("Cannot contaminate pipe ", p);
+                        LOG.error("Could not close contaminated pipe ", p);
                     }
                     if (e instanceof InterruptedException) {
                         Thread.currentThread().interrupt();
@@ -107,7 +107,7 @@ private void scheduleWorkers(CopyContext context, DataWorkerExecutor<Boolean> wo
                 context.getMigrationContext().getDataSourceRepository());
         String table = copyItem.getSourceItem();
         long totalRows = copyItem.getRowCount();
-        long pageSize = getReaderBatchSizeForTable(context, table);
+        int pageSize = copyItem.getBatchSize();
         try {
             PerformanceRecorder recorder = context.getPerformanceProfiler().createRecorder(PerformanceCategory.DB_READ,
                     table);
@@ -143,25 +143,29 @@ private void scheduleWorkers(CopyContext context, DataWorkerExecutor<Boolean> wo
                     taskRepository.updateTaskCopyMethod(context, copyItem, DataCopyMethod.OFFSET.toString());
                     taskRepository.updateTaskKeyColumns(context, copyItem, batchColumns);
 
-                    List<Long> batches = null;
+                    List<Pair<Long, Long>> batches;
                     if (context.getMigrationContext().isSchedulerResumeEnabled()) {
                         Set<DatabaseCopyBatch> pendingBatchesForPipeline = taskRepository
                                 .findPendingBatchesForPipeline(context, copyItem);
                         batches = pendingBatchesForPipeline.stream()
-                                .map(b -> Long.valueOf(b.getLowerBoundary().toString())).collect(Collectors.toList());
+                                .map(b -> Pair.of(Long.valueOf(b.getLowerBoundary().toString()),
+                                        Long.valueOf(b.getUpperBoundary().toString())))
+                                .collect(Collectors.toList());
                         taskRepository.resetPipelineBatches(context, copyItem);
                     } else {
                         batches = new ArrayList<>();
                         for (long offset = 0; offset < totalRows; offset += pageSize) {
-                            batches.add(offset);
+                            batches.add(Pair.of(offset, offset + pageSize));
                         }
                     }
 
+                    Pair<Long, Long> boundaries;
                     for (int batchId = 0; batchId < batches.size(); batchId++) {
-                        long offset = batches.get(batchId);
-                        DataReaderTask dataReaderTask = new BatchOffsetDataReaderTask(pipeTaskContext, batchId, offset,
-                                batchColumns);
-                        taskRepository.scheduleBatch(context, copyItem, batchId, offset, offset + pageSize);
+                        boundaries = batches.get(batchId);
+                        DataReaderTask dataReaderTask = new BatchOffsetDataReaderTask(pipeTaskContext, batchId,
+                                boundaries.getLeft(), batchColumns);
+                        taskRepository.scheduleBatch(context, copyItem, batchId, boundaries.getLeft(),
+                                boundaries.getRight());
                         workerExecutor.safelyExecute(dataReaderTask);
                     }
                 } else {
@@ -182,13 +186,12 @@ private void scheduleWorkers(CopyContext context, DataWorkerExecutor<Boolean> wo
                 taskRepository.updateTaskCopyMethod(context, copyItem, DataCopyMethod.SEEK.toString());
                 taskRepository.updateTaskKeyColumns(context, copyItem, Lists.newArrayList(batchColumn));
 
-                List<List<Object>> batchMarkersList = null;
+                List<List<Object>> batchMarkersList;
                 if (context.getMigrationContext().isSchedulerResumeEnabled()) {
-                    batchMarkersList = new ArrayList<>();
                     Set<DatabaseCopyBatch> pendingBatchesForPipeline = taskRepository
                             .findPendingBatchesForPipeline(context, copyItem);
-                    batchMarkersList.addAll(pendingBatchesForPipeline.stream()
-                            .map(b -> Collections.list(b.getLowerBoundary())).collect(Collectors.toList()));
+                    batchMarkersList = pendingBatchesForPipeline.stream()
+                            .map(b -> Collections.list(b.getLowerBoundary())).collect(Collectors.toList());
                     taskRepository.resetPipelineBatches(context, copyItem);
                 } else {
                     MarkersQueryDefinition queryDefinition = new MarkersQueryDefinition();
@@ -237,9 +240,4 @@ private void scheduleWorkers(CopyContext context, DataWorkerExecutor<Boolean> wo
             throw new RuntimeException("Exception while preparing reader tasks", ex);
         }
     }
-
-    private static int getReaderBatchSizeForTable(final CopyContext context, final String tableName) {
-        Integer tableBatchSize = context.getMigrationContext().getReaderBatchSize(tableName);
-        return tableBatchSize == null ? context.getMigrationContext().getReaderBatchSize() : tableBatchSize;
-    }
 }

commercedbsync/src/com/sap/cx/boosters/commercedbsync/concurrent/impl/DefaultDataWorkerExecutor.java

@@ -47,7 +47,7 @@ private Future<T> internalSafelyExecute(Callable<T> callable, int rejections) th
             for (int i = 0; i < rejections; i++) {
                 waitInterval = backOff.nextBackOff();
             }
-            LOG.trace("worker rejected. Retrying in {}ms...", waitInterval);
+            LOG.trace("Could not fetch new worker, because all are busy. Retrying again in {}ms...", waitInterval);
             Thread.sleep(waitInterval);
             return internalSafelyExecute(callable, rejections + 1);
         }

commercedbsync/src/com/sap/cx/boosters/commercedbsync/context/CopyContext.java

@@ -70,16 +70,20 @@ public static class DataCopyItem {
         private final String targetItem;
         private final Map<String, String> columnMap = new TreeMap<>(String.CASE_INSENSITIVE_ORDER);
         private final Long rowCount;
+        private final Integer batchSize;
 
-        public DataCopyItem(String sourceItem, String targetItem) {
+        public DataCopyItem(String sourceItem, String targetItem, Integer batchSize) {
             this.sourceItem = sourceItem;
             this.targetItem = targetItem;
+            this.batchSize = batchSize;
             this.rowCount = null;
         }
 
-        public DataCopyItem(String sourceItem, String targetItem, Map<String, String> columnMap, Long rowCount) {
+        public DataCopyItem(String sourceItem, String targetItem, Map<String, String> columnMap, Long rowCount,
+                Integer batchSize) {
             this.sourceItem = sourceItem;
             this.targetItem = targetItem;
+            this.batchSize = batchSize;
             this.columnMap.clear();
             this.columnMap.putAll(columnMap);
             this.rowCount = rowCount;
@@ -109,6 +113,10 @@ public Long getRowCount() {
             return rowCount;
         }
 
+        public Integer getBatchSize() {
+            return batchSize;
+        }
+
         @Override
         public String toString() {
             return new StringJoiner(", ", DataCopyItem.class.getSimpleName() + "[", "]")

commercedbsync/src/com/sap/cx/boosters/commercedbsync/context/impl/DefaultMigrationContext.java

@@ -116,7 +116,7 @@ public int getReaderBatchSize() {
     public Integer getReaderBatchSize(final String tableName) {
         String tblConfKey = CommercedbsyncConstants.MIGRATION_DATA_READER_BATCHSIZE_FOR_TABLE.replace("{table}",
                 tableName);
-        return configuration.getInteger(tblConfKey, null);
+        return configuration.getInteger(tblConfKey, getReaderBatchSize());
     }
 
     @Override

commercedbsync/src/com/sap/cx/boosters/commercedbsync/events/handlers/CopyDatabaseTableEventListener.java

@@ -56,9 +56,9 @@ protected void onEvent(final CopyDatabaseTableEvent event) {
             CopyContext copyContext = new CopyContext(migrationId, migrationContext, new HashSet<>(),
                     performanceProfiler);
             Set<DatabaseCopyTask> copyTableTasks = databaseCopyTaskRepository.findPendingTasks(copyContext);
-            Set<CopyContext.DataCopyItem> items = copyTableTasks
-                    .stream().map(task -> new CopyContext.DataCopyItem(task.getSourcetablename(),
-                            task.getTargettablename(), task.getColumnmap(), task.getSourcerowcount()))
+            Set<CopyContext.DataCopyItem> items = copyTableTasks.stream()
+                    .map(task -> new CopyContext.DataCopyItem(task.getSourcetablename(), task.getTargettablename(),
+                            task.getColumnmap(), task.getSourcerowcount(), task.getBatchsize()))
                     .collect(Collectors.toSet());
             copyContext.getCopyItems().addAll(items);
             databaseMigrationCopyService.copyAllAsync(copyContext);

commercedbsync/src/com/sap/cx/boosters/commercedbsync/jobs/AbstractMigrationJobPerformable.java

@@ -74,12 +74,41 @@ public abstract class AbstractMigrationJobPerformable extends AbstractJobPerform
     @Override
     public boolean isPerformable() {
         for (CronJobModel cronJob : getCronJobService().getRunningOrRestartedCronJobs()) {
+            currentMigrationId = databaseMigrationService.getMigrationID(migrationContext);
+
             if ((cronJob instanceof IncrementalMigrationCronJobModel || cronJob instanceof FullMigrationCronJobModel)) {
-                LOG.info("Previous migrations job already running {} and Type {} ", cronJob.getCode(),
-                        cronJob.getItemtype());
+                if (isJobStateAborted(cronJob)) {
+                    try {
+                        abortCurrentMigration();
+                        databaseMigrationService.markRemainingTasksAborted(migrationContext, currentMigrationId);
+                        clearAbortRequestedIfNeeded(cronJob);
+
+                        LOG.info("Migration with ID: {} was marked as aborted", currentMigrationId);
+                    } catch (Exception e) {
+                        LOG.warn("Failed to abort current migration");
+                        LOG.debug("Migration abort failed", e);
+                    }
+                } else {
+                    LOG.info("Previous migration job already running {} and type {}", cronJob.getCode(),
+                            cronJob.getItemtype());
+                }
                 return false;
             }
         }
+
+        if (StringUtils.isNotEmpty(currentMigrationId)) {
+            try {
+                if (databaseMigrationService.getMigrationState(migrationContext, currentMigrationId)
+                        .getStatus() == MigrationProgress.RUNNING) {
+                    LOG.info("Previous migration already running, ID: {}", currentMigrationId);
+                    return false;
+                }
+            } catch (Exception e) {
+                LOG.warn("Unable to fetch current migration status");
+                LOG.debug("Migration status fetch failed", e);
+            }
+        }
+
         return true;
     }
 
@@ -199,13 +228,12 @@ protected MigrationStatus waitForFinishCronjobs(IncrementalMigrationContext cont
         } while (StringUtils.equalsAnyIgnoreCase(status.getStatus().toString(), RUNNING_MIGRATION));
 
         if (aborted) {
-            LOG.info(" Aborted ...STOPPING migration ");
-            databaseMigrationService.stopMigration(migrationContext, currentMigrationId);
+            abortCurrentMigration();
+            clearAbortRequestedIfNeeded(cronJobModel);
             LOG.error("Database migration has been ABORTED, Migration State= " + status + ", Total Tasks "
                     + status.getTotalTasks() + ", migration id =" + status.getMigrationID() + ", Completed Tasks "
                     + status.getCompletedTasks());
-            clearAbortRequestedIfNeeded(cronJobModel);
-            throw new AbortCronJobException("CronJOB ABORTED");
+            throw new AbortCronJobException("Cronjob ABORTED");
         }
 
         if (status.isFailed()) {
@@ -218,6 +246,11 @@ protected MigrationStatus waitForFinishCronjobs(IncrementalMigrationContext cont
         return status;
     }
 
+    private void abortCurrentMigration() throws Exception {
+        LOG.info("Aborted ...STOPPING migration");
+        databaseMigrationService.stopMigration(migrationContext, currentMigrationId);
+    }
+
     protected LaunchOptions createLaunchOptions(MigrationCronJobModel migrationCronJob) {
         final LaunchOptions launchOptions = new LaunchOptions();
 
@@ -240,8 +273,8 @@ private void putLaunchOptionProperty(final LaunchOptions launchOptions, String p
 
     protected boolean isJobStateAborted(final CronJobModel cronJobModel) {
         this.modelService.refresh(cronJobModel);
-        LOG.info("cron job status = " + cronJobModel.getStatus());
-        LOG.info("cron job request to abort =" + cronJobModel.getRequestAbort());
+        LOG.info("Cron job status: {}", cronJobModel.getStatus());
+        LOG.info("Cron job request to abort: {}", BooleanUtils.isTrue(cronJobModel.getRequestAbort()));
         return ((cronJobModel.getStatus() == CronJobStatus.ABORTED)
                 || (cronJobModel.getRequestAbort() != null && cronJobModel.getRequestAbort()));
     }

commercedbsync/src/com/sap/cx/boosters/commercedbsync/jobs/FullMigrationJob.java

@@ -7,13 +7,16 @@
 package com.sap.cx.boosters.commercedbsync.jobs;
 
 import com.google.common.base.Preconditions;
+import com.sap.cx.boosters.commercedbsync.constants.CommercedbsyncConstants;
 import com.sap.cx.boosters.commercedbsync.context.IncrementalMigrationContext;
+import com.sap.cx.boosters.commercedbsync.context.LaunchOptions;
 import de.hybris.platform.cronjob.enums.CronJobResult;
 import de.hybris.platform.cronjob.enums.CronJobStatus;
 import de.hybris.platform.cronjob.jalo.AbortCronJobException;
 import de.hybris.platform.cronjob.model.CronJobModel;
 import de.hybris.platform.servicelayer.cronjob.PerformResult;
 import com.sap.cx.boosters.commercedbsync.model.cron.FullMigrationCronJobModel;
+import de.hybris.platform.servicelayer.model.ModelService;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -24,6 +27,12 @@ public class FullMigrationJob extends AbstractMigrationJobPerformable {
 
     private static final Logger LOG = LoggerFactory.getLogger(FullMigrationJob.class);
 
+    private final ModelService modelService;
+
+    public FullMigrationJob(final ModelService modelService) {
+        this.modelService = modelService;
+    }
+
     @Override
     public PerformResult perform(final CronJobModel cronJobModel) {
         FullMigrationCronJobModel fullMigrationCronJobModel;
@@ -54,9 +63,24 @@ public PerformResult perform(final CronJobModel cronJobModel) {
             incrementalMigrationContext.setIncrementalModeEnabled(false);
             incrementalMigrationContext
                     .setFullDatabaseMigrationEnabled(fullMigrationCronJobModel.isFullDatabaseMigration());
-            currentMigrationId = databaseMigrationService.startMigration(incrementalMigrationContext,
-                    createLaunchOptions(fullMigrationCronJobModel));
+            final LaunchOptions launchOptions = createLaunchOptions(fullMigrationCronJobModel);
 
+            if (fullMigrationCronJobModel.isResumeMigration()) {
+                currentMigrationId = fullMigrationCronJobModel.getMigrationId();
+                Preconditions.checkNotNull(currentMigrationId,
+                        "Migration ID must be present to resume failed migration job");
+                launchOptions.getPropertyOverrideMap().put(CommercedbsyncConstants.MIGRATION_SCHEDULER_RESUME_ENABLED,
+                        true);
+                databaseMigrationService.resumeUnfinishedMigration(incrementalMigrationContext, launchOptions,
+                        currentMigrationId);
+                LOG.info("Resumed Migration {}", currentMigrationId);
+            } else {
+                currentMigrationId = databaseMigrationService.startMigration(incrementalMigrationContext,
+                        launchOptions);
+                LOG.info("Started Migration {}", currentMigrationId);
+                fullMigrationCronJobModel.setMigrationId(currentMigrationId);
+                modelService.save(fullMigrationCronJobModel);
+            }
             waitForFinishCronjobs(incrementalMigrationContext, currentMigrationId, cronJobModel);
         } catch (final AbortCronJobException e) {
             return new PerformResult(CronJobResult.ERROR, CronJobStatus.ABORTED);