InventoryDataScalingTask extends AbstractShardingScalingExecutor<InventoryPosition> implements ScalingTask<InventoryPosition> { @Override public void start() { getEstimatedRows(); instanceDumper(); Importer importer = ImporterFactory.newInstance(importerConfig, dataSourceManager); instanceChannel(importer); Future<?> future = ScalingContext.getInstance().getImporterExecuteEngine().submit(importer, new ExecuteCallback() { @Override public void onSuccess() { } @Override public void onFailure(final Throwable throwable) { log.error("get an error when migrating the inventory data", throwable); dumper.stop(); } }); dumper.start(); waitForResult(future); dataSourceManager.close(); } InventoryDataScalingTask(final InventoryDumperConfiguration inventoryDumperConfig, final ImporterConfiguration importerConfig); @SuppressWarnings("unchecked")  InventoryDataScalingTask(final InventoryDumperConfiguration inventoryDumperConfig, final ImporterConfiguration importerConfig, final DataSourceManager dataSourceManager); @Override void start(); @Override void stop(); @Override SyncProgress getProgress();  }
@Test(expected = SyncTaskExecuteException.class) public void assertStartWithGetEstimatedRowsFailure() { InventoryDumperConfiguration inventoryDumperConfig = new InventoryDumperConfiguration(syncConfig.getDumperConfiguration()); inventoryDumperConfig.setTableName("t_non_exist"); InventoryDataScalingTask inventoryDataSyncTask = new InventoryDataScalingTask(inventoryDumperConfig, syncConfig.getImporterConfiguration(), dataSourceManager); inventoryDataSyncTask.start(); }