@Test public void cleanupDropsCheckpointTable() { fromKvs.createTables(TEST_AND_CHECKPOINT_TABLES); fromTxManager.runTaskWithRetry(tx -> { tx.put(TEST_TABLE, ImmutableMap.of(TEST_CELL, TEST_VALUE1)); return tx.getTimestamp(); }); KeyValueServiceMigrator migrator = KeyValueServiceMigrators.setupMigrator(migratorSpec); migrator.setup(); migrator.migrate(); verify(toKvs, never()).dropTable(CHECKPOINT_TABLE); migrator.cleanup(); verify(toKvs, times(1)).dropTable(CHECKPOINT_TABLE); }
public static KeyValueServiceMigrator setupMigrator(MigratorSpec migratorSpec) { AtlasDbServices fromServices = migratorSpec.fromServices(); AtlasDbServices toServices = migratorSpec.toServices(); TimestampManagementService toTimestampManagementService = toServices.getManagedTimestampService(); toTimestampManagementService.fastForwardTimestamp( fromServices.getManagedTimestampService().getFreshTimestamp() + 1); long migrationStartTimestamp = toServices.getManagedTimestampService().getFreshTimestamp(); long migrationCommitTimestamp = toServices.getManagedTimestampService().getFreshTimestamp(); toServices.getTransactionService().putUnlessExists(migrationStartTimestamp, migrationCommitTimestamp); return new KeyValueServiceMigrator( CHECKPOINT_NAMESPACE, fromServices.getTransactionManager(), toServices.getTransactionManager(), fromServices.getKeyValueService(), toServices.getKeyValueService(), Suppliers.ofInstance(migrationStartTimestamp), migratorSpec.threads(), migratorSpec.batchSize(), ImmutableMap.of(), (String message, KeyValueServiceMigrator.KvsMigrationMessageLevel level) -> printer.info(level.toString() + ": " + message), new TaskProgress() { @Override public void beginTask(String message, int tasks) { printer.info(message); } @Override public void subTaskComplete() { } @Override public void taskComplete() { } }, ImmutableSet.of()); }
KeyValueServiceMigrators { public static KeyValueServiceMigrator setupMigrator(MigratorSpec migratorSpec) { AtlasDbServices fromServices = migratorSpec.fromServices(); AtlasDbServices toServices = migratorSpec.toServices(); TimestampManagementService toTimestampManagementService = toServices.getManagedTimestampService(); toTimestampManagementService.fastForwardTimestamp( fromServices.getManagedTimestampService().getFreshTimestamp() + 1); long migrationStartTimestamp = toServices.getManagedTimestampService().getFreshTimestamp(); long migrationCommitTimestamp = toServices.getManagedTimestampService().getFreshTimestamp(); toServices.getTransactionService().putUnlessExists(migrationStartTimestamp, migrationCommitTimestamp); return new KeyValueServiceMigrator( CHECKPOINT_NAMESPACE, fromServices.getTransactionManager(), toServices.getTransactionManager(), fromServices.getKeyValueService(), toServices.getKeyValueService(), Suppliers.ofInstance(migrationStartTimestamp), migratorSpec.threads(), migratorSpec.batchSize(), ImmutableMap.of(), (String message, KeyValueServiceMigrator.KvsMigrationMessageLevel level) -> printer.info(level.toString() + ": " + message), new TaskProgress() { @Override public void beginTask(String message, int tasks) { printer.info(message); } @Override public void subTaskComplete() { } @Override public void taskComplete() { } }, ImmutableSet.of()); } }
KeyValueServiceMigrators { public static KeyValueServiceMigrator setupMigrator(MigratorSpec migratorSpec) { AtlasDbServices fromServices = migratorSpec.fromServices(); AtlasDbServices toServices = migratorSpec.toServices(); TimestampManagementService toTimestampManagementService = toServices.getManagedTimestampService(); toTimestampManagementService.fastForwardTimestamp( fromServices.getManagedTimestampService().getFreshTimestamp() + 1); long migrationStartTimestamp = toServices.getManagedTimestampService().getFreshTimestamp(); long migrationCommitTimestamp = toServices.getManagedTimestampService().getFreshTimestamp(); toServices.getTransactionService().putUnlessExists(migrationStartTimestamp, migrationCommitTimestamp); return new KeyValueServiceMigrator( CHECKPOINT_NAMESPACE, fromServices.getTransactionManager(), toServices.getTransactionManager(), fromServices.getKeyValueService(), toServices.getKeyValueService(), Suppliers.ofInstance(migrationStartTimestamp), migratorSpec.threads(), migratorSpec.batchSize(), ImmutableMap.of(), (String message, KeyValueServiceMigrator.KvsMigrationMessageLevel level) -> printer.info(level.toString() + ": " + message), new TaskProgress() { @Override public void beginTask(String message, int tasks) { printer.info(message); } @Override public void subTaskComplete() { } @Override public void taskComplete() { } }, ImmutableSet.of()); } private  KeyValueServiceMigrators(); }
KeyValueServiceMigrators { public static KeyValueServiceMigrator setupMigrator(MigratorSpec migratorSpec) { AtlasDbServices fromServices = migratorSpec.fromServices(); AtlasDbServices toServices = migratorSpec.toServices(); TimestampManagementService toTimestampManagementService = toServices.getManagedTimestampService(); toTimestampManagementService.fastForwardTimestamp( fromServices.getManagedTimestampService().getFreshTimestamp() + 1); long migrationStartTimestamp = toServices.getManagedTimestampService().getFreshTimestamp(); long migrationCommitTimestamp = toServices.getManagedTimestampService().getFreshTimestamp(); toServices.getTransactionService().putUnlessExists(migrationStartTimestamp, migrationCommitTimestamp); return new KeyValueServiceMigrator( CHECKPOINT_NAMESPACE, fromServices.getTransactionManager(), toServices.getTransactionManager(), fromServices.getKeyValueService(), toServices.getKeyValueService(), Suppliers.ofInstance(migrationStartTimestamp), migratorSpec.threads(), migratorSpec.batchSize(), ImmutableMap.of(), (String message, KeyValueServiceMigrator.KvsMigrationMessageLevel level) -> printer.info(level.toString() + ": " + message), new TaskProgress() { @Override public void beginTask(String message, int tasks) { printer.info(message); } @Override public void subTaskComplete() { } @Override public void taskComplete() { } }, ImmutableSet.of()); } private  KeyValueServiceMigrators(); static KeyValueServiceMigrator setupMigrator(MigratorSpec migratorSpec); }
KeyValueServiceMigrators { public static KeyValueServiceMigrator setupMigrator(MigratorSpec migratorSpec) { AtlasDbServices fromServices = migratorSpec.fromServices(); AtlasDbServices toServices = migratorSpec.toServices(); TimestampManagementService toTimestampManagementService = toServices.getManagedTimestampService(); toTimestampManagementService.fastForwardTimestamp( fromServices.getManagedTimestampService().getFreshTimestamp() + 1); long migrationStartTimestamp = toServices.getManagedTimestampService().getFreshTimestamp(); long migrationCommitTimestamp = toServices.getManagedTimestampService().getFreshTimestamp(); toServices.getTransactionService().putUnlessExists(migrationStartTimestamp, migrationCommitTimestamp); return new KeyValueServiceMigrator( CHECKPOINT_NAMESPACE, fromServices.getTransactionManager(), toServices.getTransactionManager(), fromServices.getKeyValueService(), toServices.getKeyValueService(), Suppliers.ofInstance(migrationStartTimestamp), migratorSpec.threads(), migratorSpec.batchSize(), ImmutableMap.of(), (String message, KeyValueServiceMigrator.KvsMigrationMessageLevel level) -> printer.info(level.toString() + ": " + message), new TaskProgress() { @Override public void beginTask(String message, int tasks) { printer.info(message); } @Override public void subTaskComplete() { } @Override public void taskComplete() { } }, ImmutableSet.of()); } private  KeyValueServiceMigrators(); static KeyValueServiceMigrator setupMigrator(MigratorSpec migratorSpec);  }