public Cancellable scheduleSecureStoreUpdate()

in twill-yarn/src/main/java/org/apache/twill/yarn/YarnTwillRunnerService.java [212:262]
42 lines of code
6 McCabe index (conditional complexity)

  public Cancellable scheduleSecureStoreUpdate(final SecureStoreUpdater updater,
                                               long initialDelay, long delay, TimeUnit unit) {
    if (!UserGroupInformation.isSecurityEnabled()) {
      return new Cancellable() {
        @Override
        public void cancel() {
          // No-op
        }
      };
    }

    synchronized (this) {
      if (secureStoreScheduler == null) {
        secureStoreScheduler = Executors.newSingleThreadScheduledExecutor(
          Threads.createDaemonThreadFactory("secure-store-updater"));
      }
    }

    final ScheduledFuture<?> future = secureStoreScheduler.scheduleWithFixedDelay(new Runnable() {
      @Override
      public void run() {
        // Collects all <application, runId> pairs first
        Multimap<String, RunId> liveApps = HashMultimap.create();
        synchronized (YarnTwillRunnerService.this) {
          for (Table.Cell<String, RunId, YarnTwillController> cell : controllers.cellSet()) {
            liveApps.put(cell.getRowKey(), cell.getColumnKey());
          }
        }

        // Collect all secure stores that needs to be updated.
        Table<String, RunId, SecureStore> secureStores = HashBasedTable.create();
        for (Map.Entry<String, RunId> entry : liveApps.entries()) {
          try {
            secureStores.put(entry.getKey(), entry.getValue(), updater.update(entry.getKey(), entry.getValue()));
          } catch (Throwable t) {
            LOG.warn("Exception thrown by SecureStoreUpdater {}", updater, t);
          }
        }

        // Update secure stores.
        updateSecureStores(secureStores);
      }
    }, initialDelay, delay, unit);

    return new Cancellable() {
      @Override
      public void cancel() {
        future.cancel(false);
      }
    };
  }