+
+ public void waitTillReady() {
+ LOG.info("Beginning to wait for data store to become ready : {}", identifier);
+
+ try {
+ if (waitTillReadyCountDownLatch.await(waitTillReadyTimeInMillis, TimeUnit.MILLISECONDS)) {
+ LOG.debug("Data store {} is now ready", identifier);
+ } else {
+ LOG.error("Shard leaders failed to settle in {} seconds, giving up",
+ TimeUnit.MILLISECONDS.toSeconds(waitTillReadyTimeInMillis));
+ }
+ } catch (InterruptedException e) {
+ LOG.error("Interrupted while waiting for shards to settle", e);
+ }
+ }
+
+ @SuppressWarnings("checkstyle:IllegalCatch")
+ private static ActorRef createShardManager(final ActorSystem actorSystem, final ShardManagerCreator creator,
+ final String shardDispatcher, final String shardManagerId) {
+ Exception lastException = null;
+
+ for (int i = 0; i < 100; i++) {
+ try {
+ return actorSystem.actorOf(creator.props().withDispatcher(shardDispatcher).withMailbox(
+ ActorContext.BOUNDED_MAILBOX), shardManagerId);
+ } catch (Exception e) {
+ lastException = e;
+ Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+ LOG.debug("Could not create actor {} because of {} - waiting for sometime before retrying "
+ + "(retry count = {})", shardManagerId, e.getMessage(), i);
+ }
+ }
+
+ throw new IllegalStateException("Failed to create Shard Manager", lastException);
+ }
+
+ @VisibleForTesting
+ public CountDownLatch getWaitTillReadyCountDownLatch() {
+ return waitTillReadyCountDownLatch;
+ }