code: @Override
    public void onNext(final StartTime startTime) {
  
      LOG.log(Level.INFO, "Driver started: app {0} :: {1}", new Object[] {this.hostApplicationId, startTime});
      LOG.log(Level.FINE, "Launching Unmanaged AM: {0}", JAR_PATH);
  
      try (final DriverLauncher client = DriverLauncher.getLauncher(RUNTIME_CONFIG)) {
  
        final String innerApplicationId = client.submit(DRIVER_CONFIG, 10000);
        LOG.log(Level.INFO, "REEF-on-REEF job submitted: host app {0} new app {1}",
            new Object[] {this.hostApplicationId, innerApplicationId});
  
        final Configuration yarnAmConfig = UnmanagedAmYarnDriverConfiguration.CONF
            .set(UnmanagedAmYarnDriverConfiguration.JOB_IDENTIFIER, innerApplicationId)
            .set(UnmanagedAmYarnDriverConfiguration.JOB_SUBMISSION_DIRECTORY, DRIVER_ROOT_PATH)
            .build();
  
        try (final REEFEnvironment reef =
            REEFEnvironment.fromConfiguration(client.getUser(), yarnAmConfig, DRIVER_CONFIG)) {
  
          reef.run();
  
          final ReefServiceProtos.JobStatusProto status = reef.getLastStatus();
  
          LOG.log(Level.INFO, "REEF-on-REEF inner job {0} completed: state {1}",
              new Object[] {innerApplicationId, status.getState()});
        }
  
        LOG.log(Level.INFO,
            "REEF-on-REEF host job {0} completed: inner app {1} status {2}",
            new Object[] {this.hostApplicationId, innerApplicationId, client.getStatus()});
  
      } catch (final InjectionException ex) {
        LOG.log(Level.SEVERE, "REEF-on-REEF configuration error", ex);
        throw new RuntimeException("REEF-on-REEF configuration error", ex);
      }
  
      ThreadLogger.logThreads(LOG, Level.FINEST, "Threads running after DriverLauncher.close():");
    }
docs: StartTime event: launch another REEF job.