@Test public void testScrubQueueMigration() { TableReference foo = TableReference.createWithEmptyNamespace("foo"); TableReference bar = TableReference.createWithEmptyNamespace("bar"); TableReference baz = TableReference.createWithEmptyNamespace("baz"); Cell cell1 = Cell.create(new byte[] {1, 2, 3}, new byte[] {4, 5, 6}); Cell cell2 = Cell.create(new byte[] {7, 8, 9}, new byte[] {4, 5, 6}); Cell cell3 = Cell.create(new byte[] {1, 2, 3}, new byte[] {7, 8, 9}); kvs.createTable(AtlasDbConstants.OLD_SCRUB_TABLE, new byte[] {0}); kvs.putWithTimestamps(AtlasDbConstants.OLD_SCRUB_TABLE, ImmutableMultimap.of( cell1, Value.create("foo\0bar".getBytes(), 10), cell1, Value.create("baz".getBytes(), 20), cell2, Value.create("foo".getBytes(), 30), cell3, Value.create("foo\0bar\0baz".getBytes(), 40))); ByteArrayOutputStream baos = new ByteArrayOutputStream(); ScrubQueueMigrationCommand.run(kvs, new PrintWriter(baos, true), 1000); BatchingVisitable<SortedMap<Long, Multimap<TableReference, Cell>>> visitable = scrubStore.getBatchingVisitableScrubQueue( Long.MAX_VALUE, PtBytes.EMPTY_BYTE_ARRAY, PtBytes.EMPTY_BYTE_ARRAY); Assert.assertEquals(ImmutableSortedMap.of( 10L, ImmutableMultimap.of(foo, cell1, bar, cell1), 20L, ImmutableMultimap.of(baz, cell1), 30L, ImmutableMultimap.of(foo, cell2), 40L, ImmutableMultimap.of(foo, cell3, bar, cell3, baz, cell3)), Iterables.getOnlyElement(BatchingVisitables.copyToList(visitable))); String output = new String(baos.toByteArray()); Assert.assertTrue(output, output.contains("Starting iteration 2")); Assert.assertTrue(output, output.contains("Moved 4 cells")); Assert.assertTrue(output, output.contains("into 7 cells")); Assert.assertTrue(output, output.contains("Finished all iterations")); }
public static void run(KeyValueService kvs, PrintWriter output, int batchSize) { Context context = new Context(kvs, output, batchSize); for (int i = 0;; i++) { output.println("Starting iteration " + i + " of scrub migration."); Stopwatch watch = Stopwatch.createStarted(); try (ClosableIterator<RowResult<Value>> iter = kvs.getRange( AtlasDbConstants.OLD_SCRUB_TABLE, RangeRequest.all(), Long.MAX_VALUE)) { if (!iter.hasNext()) { output.println("Finished all iterations of scrub migration."); break; } runOnce(context, iter); } output.println("Finished iteration " + i + " of scrub migration in " + watch.elapsed(TimeUnit.SECONDS) + " seconds."); } }
ScrubQueueMigrationCommand extends SingleBackendCommand { public static void run(KeyValueService kvs, PrintWriter output, int batchSize) { Context context = new Context(kvs, output, batchSize); for (int i = 0;; i++) { output.println("Starting iteration " + i + " of scrub migration."); Stopwatch watch = Stopwatch.createStarted(); try (ClosableIterator<RowResult<Value>> iter = kvs.getRange( AtlasDbConstants.OLD_SCRUB_TABLE, RangeRequest.all(), Long.MAX_VALUE)) { if (!iter.hasNext()) { output.println("Finished all iterations of scrub migration."); break; } runOnce(context, iter); } output.println("Finished iteration " + i + " of scrub migration in " + watch.elapsed(TimeUnit.SECONDS) + " seconds."); } } }
ScrubQueueMigrationCommand extends SingleBackendCommand { public static void run(KeyValueService kvs, PrintWriter output, int batchSize) { Context context = new Context(kvs, output, batchSize); for (int i = 0;; i++) { output.println("Starting iteration " + i + " of scrub migration."); Stopwatch watch = Stopwatch.createStarted(); try (ClosableIterator<RowResult<Value>> iter = kvs.getRange( AtlasDbConstants.OLD_SCRUB_TABLE, RangeRequest.all(), Long.MAX_VALUE)) { if (!iter.hasNext()) { output.println("Finished all iterations of scrub migration."); break; } runOnce(context, iter); } output.println("Finished iteration " + i + " of scrub migration in " + watch.elapsed(TimeUnit.SECONDS) + " seconds."); } }  }
ScrubQueueMigrationCommand extends SingleBackendCommand { public static void run(KeyValueService kvs, PrintWriter output, int batchSize) { Context context = new Context(kvs, output, batchSize); for (int i = 0;; i++) { output.println("Starting iteration " + i + " of scrub migration."); Stopwatch watch = Stopwatch.createStarted(); try (ClosableIterator<RowResult<Value>> iter = kvs.getRange( AtlasDbConstants.OLD_SCRUB_TABLE, RangeRequest.all(), Long.MAX_VALUE)) { if (!iter.hasNext()) { output.println("Finished all iterations of scrub migration."); break; } runOnce(context, iter); } output.println("Finished iteration " + i + " of scrub migration in " + watch.elapsed(TimeUnit.SECONDS) + " seconds."); } }  @Override int execute(AtlasDbServices services); static void run(KeyValueService kvs,
                           PrintWriter output,
                           int batchSize); @Override boolean isOnlineRunSupported(); }
ScrubQueueMigrationCommand extends SingleBackendCommand { public static void run(KeyValueService kvs, PrintWriter output, int batchSize) { Context context = new Context(kvs, output, batchSize); for (int i = 0;; i++) { output.println("Starting iteration " + i + " of scrub migration."); Stopwatch watch = Stopwatch.createStarted(); try (ClosableIterator<RowResult<Value>> iter = kvs.getRange( AtlasDbConstants.OLD_SCRUB_TABLE, RangeRequest.all(), Long.MAX_VALUE)) { if (!iter.hasNext()) { output.println("Finished all iterations of scrub migration."); break; } runOnce(context, iter); } output.println("Finished iteration " + i + " of scrub migration in " + watch.elapsed(TimeUnit.SECONDS) + " seconds."); } }  @Override int execute(AtlasDbServices services); static void run(KeyValueService kvs,
                           PrintWriter output,
                           int batchSize); @Override boolean isOnlineRunSupported();  }