@Test public void doesNotInvokeBackoffIfCalledWithZeroTime() { StreamStoreBackoffStrategy backoffStrategy = createStaticBackoffStrategy(1, 0); backoffStrategy.accept(1); verifyNoMoreInteractions(backoffMechanism); }
@Override public void accept(long blockNumber) { if (persistenceConfiguration.get().writePauseDurationMillis() == 0) { return; } if (shouldBackoffBeforeWritingBlockNumber(blockNumber)) { try { long writePauseDurationMillis = persistenceConfiguration.get().writePauseDurationMillis(); log.info("Invoking backoff for {} ms, because we are writing block {} of a stream", SafeArg.of("blockNumber", blockNumber), SafeArg.of("writePauseDurationMillis", writePauseDurationMillis)); backoff.backoff(writePauseDurationMillis); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } }
StandardPeriodicBackoffStrategy implements StreamStoreBackoffStrategy { @Override public void accept(long blockNumber) { if (persistenceConfiguration.get().writePauseDurationMillis() == 0) { return; } if (shouldBackoffBeforeWritingBlockNumber(blockNumber)) { try { long writePauseDurationMillis = persistenceConfiguration.get().writePauseDurationMillis(); log.info("Invoking backoff for {} ms, because we are writing block {} of a stream", SafeArg.of("blockNumber", blockNumber), SafeArg.of("writePauseDurationMillis", writePauseDurationMillis)); backoff.backoff(writePauseDurationMillis); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } } }
StandardPeriodicBackoffStrategy implements StreamStoreBackoffStrategy { @Override public void accept(long blockNumber) { if (persistenceConfiguration.get().writePauseDurationMillis() == 0) { return; } if (shouldBackoffBeforeWritingBlockNumber(blockNumber)) { try { long writePauseDurationMillis = persistenceConfiguration.get().writePauseDurationMillis(); log.info("Invoking backoff for {} ms, because we are writing block {} of a stream", SafeArg.of("blockNumber", blockNumber), SafeArg.of("writePauseDurationMillis", writePauseDurationMillis)); backoff.backoff(writePauseDurationMillis); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } } @VisibleForTesting  StandardPeriodicBackoffStrategy(Supplier<StreamStorePersistenceConfiguration> persistenceConfiguration,
            BackoffMechanism backoff); }
StandardPeriodicBackoffStrategy implements StreamStoreBackoffStrategy { @Override public void accept(long blockNumber) { if (persistenceConfiguration.get().writePauseDurationMillis() == 0) { return; } if (shouldBackoffBeforeWritingBlockNumber(blockNumber)) { try { long writePauseDurationMillis = persistenceConfiguration.get().writePauseDurationMillis(); log.info("Invoking backoff for {} ms, because we are writing block {} of a stream", SafeArg.of("blockNumber", blockNumber), SafeArg.of("writePauseDurationMillis", writePauseDurationMillis)); backoff.backoff(writePauseDurationMillis); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } } @VisibleForTesting  StandardPeriodicBackoffStrategy(Supplier<StreamStorePersistenceConfiguration> persistenceConfiguration,
            BackoffMechanism backoff); static StandardPeriodicBackoffStrategy create(
            Supplier<StreamStorePersistenceConfiguration> persistenceConfiguration); @Override void accept(long blockNumber); }
StandardPeriodicBackoffStrategy implements StreamStoreBackoffStrategy { @Override public void accept(long blockNumber) { if (persistenceConfiguration.get().writePauseDurationMillis() == 0) { return; } if (shouldBackoffBeforeWritingBlockNumber(blockNumber)) { try { long writePauseDurationMillis = persistenceConfiguration.get().writePauseDurationMillis(); log.info("Invoking backoff for {} ms, because we are writing block {} of a stream", SafeArg.of("blockNumber", blockNumber), SafeArg.of("writePauseDurationMillis", writePauseDurationMillis)); backoff.backoff(writePauseDurationMillis); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } } @VisibleForTesting  StandardPeriodicBackoffStrategy(Supplier<StreamStorePersistenceConfiguration> persistenceConfiguration,
            BackoffMechanism backoff); static StandardPeriodicBackoffStrategy create(
            Supplier<StreamStorePersistenceConfiguration> persistenceConfiguration); @Override void accept(long blockNumber);  }