@Test public void testProgress() throws Exception { byte[] partitionBytes = "partition1".getBytes(StandardCharsets.UTF_8); PartitionName partitionName = new PartitionName(false, partitionBytes, partitionBytes); RingMember ringMember = new RingMember("member1"); TimestampedOrderIdProvider orderIdProvider = new OrderIdProviderImpl(new ConstantWriterIdProvider(1), new SnowflakeIdPacker(), new JiveEpochTimestampProvider()); PartitionClientProvider partitionClientProvider = new InMemoryPartitionClientProvider(orderIdProvider, ringMember); PartitionClient partition = partitionClientProvider.getPartition(partitionName); AmzaClientAquariumProvider amzaClientAquariumProvider = new AmzaClientAquariumProvider(new AquariumStats(), "test", partitionClientProvider, orderIdProvider, ringMember.asAquariumMember(), count -> count == 1, () -> Sets.newHashSet(ringMember.asAquariumMember()), 128, 128, 5_000L, 100L, 60_000L, 10_000L, Executors.newSingleThreadExecutor(), 100L, 1_000L, 10_000L, false); int[] rowCount = new int[1]; AmzaSyncClient syncClient = new AmzaSyncClient() { @Override public void commitRows(PartitionName toPartitionName, List<Row> rows) throws Exception { rowCount[0] += rows.size(); } @Override public void ensurePartition(PartitionName toPartitionName, PartitionProperties properties, int ringSize) throws Exception { } }; AmzaSyncSender syncSender = new AmzaSyncSender( new AmzaSyncStats(), new AmzaSyncSenderConfig("default", true, 100L, 1_000, false, "", "", -1, "", "", "", true), amzaClientAquariumProvider, 1, Executors.newScheduledThreadPool(1), partitionClientProvider, syncClient, (name) -> ImmutableMap.of(new AmzaSyncPartitionTuple(partitionName, partitionName), new AmzaSyncPartitionConfig(0, 0, 0, 0, 0)), new AmzaInterner()); amzaClientAquariumProvider.start(); syncSender.start(); AtomicInteger keyProvider = new AtomicInteger(); AtomicLong valueProvider = new AtomicLong(); AtomicLong largestTxId = new AtomicLong(); long initialTimeMillis = System.currentTimeMillis(); largestTxId.set(advancePartition(partition, 10, initialTimeMillis, keyProvider, valueProvider)); Assert.assertTrue(largestTxId.get() > 0, "Expected positive txId"); long failAfter = System.currentTimeMillis() + 60_000L; Cursor cursor = awaitCursor(partitionName, syncSender, ringMember, largestTxId.get(), failAfter); Assert.assertNotNull(cursor); Assert.assertFalse(cursor.taking); Assert.assertEquals(initialTimeMillis, cursor.maxTimestamp); Assert.assertNotNull(cursor.memberTxIds); Assert.assertTrue(cursor.memberTxIds.containsKey(ringMember)); Assert.assertEquals(cursor.memberTxIds.get(ringMember).longValue(), 10L); Assert.assertEquals(rowCount[0], 10); rowCount[0] = 0; long nextTimeMillis = initialTimeMillis + 1; largestTxId.set(advancePartition(partition, 10, nextTimeMillis, keyProvider, valueProvider)); Assert.assertTrue(largestTxId.get() > 0, "Expected positive txId"); cursor = awaitCursor(partitionName, syncSender, ringMember, largestTxId.get(), failAfter); Assert.assertNotNull(cursor); Assert.assertFalse(cursor.taking); Assert.assertEquals(nextTimeMillis, cursor.maxTimestamp); Assert.assertNotNull(cursor.memberTxIds); Assert.assertTrue(cursor.memberTxIds.containsKey(ringMember)); Assert.assertEquals(cursor.memberTxIds.get(ringMember).longValue(), 20L); Assert.assertEquals(rowCount[0], 10); }
public void start() { if (running.compareAndSet(false, true)) { for (int i = 0; i < syncRingStripes; i++) { amzaClientAquariumProvider.register(aquariumName(i)); } for (int i = 0; i < syncRingStripes; i++) { int index = i; syncFutures[i] = executorService.scheduleWithFixedDelay(() -> { try { syncStripe(index); } catch (InterruptedException e) { LOG.info("Sync thread {} was interrupted", index); } catch (Throwable t) { LOG.error("Failure in sync thread {}", new Object[] { index }, t); } }, 0, config.syncIntervalMillis, TimeUnit.MILLISECONDS); } } }
AmzaSyncSender { public void start() { if (running.compareAndSet(false, true)) { for (int i = 0; i < syncRingStripes; i++) { amzaClientAquariumProvider.register(aquariumName(i)); } for (int i = 0; i < syncRingStripes; i++) { int index = i; syncFutures[i] = executorService.scheduleWithFixedDelay(() -> { try { syncStripe(index); } catch (InterruptedException e) { LOG.info("Sync thread {} was interrupted", index); } catch (Throwable t) { LOG.error("Failure in sync thread {}", new Object[] { index }, t); } }, 0, config.syncIntervalMillis, TimeUnit.MILLISECONDS); } } } }
AmzaSyncSender { public void start() { if (running.compareAndSet(false, true)) { for (int i = 0; i < syncRingStripes; i++) { amzaClientAquariumProvider.register(aquariumName(i)); } for (int i = 0; i < syncRingStripes; i++) { int index = i; syncFutures[i] = executorService.scheduleWithFixedDelay(() -> { try { syncStripe(index); } catch (InterruptedException e) { LOG.info("Sync thread {} was interrupted", index); } catch (Throwable t) { LOG.error("Failure in sync thread {}", new Object[] { index }, t); } }, 0, config.syncIntervalMillis, TimeUnit.MILLISECONDS); } } } AmzaSyncSender(AmzaSyncStats stats,
        AmzaSyncSenderConfig config,
        AmzaClientAquariumProvider amzaClientAquariumProvider,
        int syncRingStripes,
        ScheduledExecutorService executorService,
        PartitionClientProvider partitionClientProvider,
        AmzaSyncClient toSyncClient,
        AmzaSyncPartitionConfigProvider syncPartitionConfigProvider,
        AmzaInterner amzaInterner); }
AmzaSyncSender { public void start() { if (running.compareAndSet(false, true)) { for (int i = 0; i < syncRingStripes; i++) { amzaClientAquariumProvider.register(aquariumName(i)); } for (int i = 0; i < syncRingStripes; i++) { int index = i; syncFutures[i] = executorService.scheduleWithFixedDelay(() -> { try { syncStripe(index); } catch (InterruptedException e) { LOG.info("Sync thread {} was interrupted", index); } catch (Throwable t) { LOG.error("Failure in sync thread {}", new Object[] { index }, t); } }, 0, config.syncIntervalMillis, TimeUnit.MILLISECONDS); } } } AmzaSyncSender(AmzaSyncStats stats,
        AmzaSyncSenderConfig config,
        AmzaClientAquariumProvider amzaClientAquariumProvider,
        int syncRingStripes,
        ScheduledExecutorService executorService,
        PartitionClientProvider partitionClientProvider,
        AmzaSyncClient toSyncClient,
        AmzaSyncPartitionConfigProvider syncPartitionConfigProvider,
        AmzaInterner amzaInterner); AmzaSyncSenderConfig getConfig(); boolean configHasChanged(AmzaSyncSenderConfig senderConfig); void start(); void stop(); void streamCursors(PartitionName fromPartitionName, PartitionName toPartitionName, ProgressStream stream); boolean resetCursors(PartitionName partitionName); }
AmzaSyncSender { public void start() { if (running.compareAndSet(false, true)) { for (int i = 0; i < syncRingStripes; i++) { amzaClientAquariumProvider.register(aquariumName(i)); } for (int i = 0; i < syncRingStripes; i++) { int index = i; syncFutures[i] = executorService.scheduleWithFixedDelay(() -> { try { syncStripe(index); } catch (InterruptedException e) { LOG.info("Sync thread {} was interrupted", index); } catch (Throwable t) { LOG.error("Failure in sync thread {}", new Object[] { index }, t); } }, 0, config.syncIntervalMillis, TimeUnit.MILLISECONDS); } } } AmzaSyncSender(AmzaSyncStats stats,
        AmzaSyncSenderConfig config,
        AmzaClientAquariumProvider amzaClientAquariumProvider,
        int syncRingStripes,
        ScheduledExecutorService executorService,
        PartitionClientProvider partitionClientProvider,
        AmzaSyncClient toSyncClient,
        AmzaSyncPartitionConfigProvider syncPartitionConfigProvider,
        AmzaInterner amzaInterner); AmzaSyncSenderConfig getConfig(); boolean configHasChanged(AmzaSyncSenderConfig senderConfig); void start(); void stop(); void streamCursors(PartitionName fromPartitionName, PartitionName toPartitionName, ProgressStream stream); boolean resetCursors(PartitionName partitionName);  }