Merge pull request #14 from SAP/develop

Fix issues with migration resuming

Author: lnowakowski

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;
-    }
 }