Java 类java.util.concurrent.ScheduledExecutorService 实例源码

public HTTPconThread(int id,URLdetails obj,int time){
    this.obj=obj;
    this.id = id;
    this.time=time;
    this.index = Controller.getList().indexOf(obj);
    final ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
    executorService.scheduleWithFixedDelay(new Runnable() {
        @Override
        public void run() {
            if(Controller.getList().indexOf(obj)==-1){
                executorService.shutdown();
            }
            testIt(obj.getUrl());
        }
    }, 0, time, TimeUnit.SECONDS);
}

@SuppressWarnings("serial")
private SucceededListener getUploadSucceededListener() {
    return new SucceededListener() {

        @Override
        public void uploadSucceeded(SucceededEvent event) {
            log.info("Upload Successful! Analyzing Uploaded Image.....");
            final ProgressIndicatorWindow progressIndicatorWindow = new ProgressIndicatorWindow();

            progressIndicatorWindow.setWidth("200px");
            progressIndicatorWindow.setHeight("100px");
            progressIndicatorWindow.setCaption("Processing image ...");

            UI.getCurrent().addWindow(progressIndicatorWindow);
            progressIndicatorWindow.bringToFront();

            Runnable serviceCall = uploadValidationService(progressIndicatorWindow, event);

            ScheduledExecutorService exec = Executors.newScheduledThreadPool(1);
            exec.schedule(serviceCall, 1, TimeUnit.MILLISECONDS);
        }
    };
}

public void init(){
    String cleanTime = logCleanTime;
    Date nextExeucteTime = calcNextExecuteTime(cleanTime);
    long initialDelay = nextExeucteTime.getTime() - System.currentTimeMillis();

    ScheduledExecutorService executor = Executors.newScheduledThreadPool(1, new NamedThreadFactory("CleanLogJob",true));
    executor.scheduleAtFixedRate(new Runnable() {
        @Override
        public void run() {
            try{
                if(master.hasLeaderShip()){
                    Calendar instance = Calendar.getInstance();
                    instance.add(Calendar.DATE, -logReservedDays);
                    LOG.info("START CLEAN EXPIRED TRANSACTION LOGS.DAYS:" + logReservedDays);
                    logWritter.cleanFinishedLogs(applicationName, instance.getTime());
                    LOG.info("END CLEAN EXPIRED TRANSACTION LOGS.DAYS");
                }else{
                    LOG.info("NOT MASTER,do not execute transaction log clean job");
                }
            }catch(Exception e){
                LOG.error("execute clean job error!",e);
            }
        }
    }, initialDelay, 24l*60*60*1000 , TimeUnit.MILLISECONDS);
}

@Override
public boolean onKeyDown(int keyCode, KeyEvent event) {
    if (keyCode == KeyEvent.KEYCODE_BACK) {
        if (!isExit) {
            isExit = true;
            Toast.makeText(this, "再按一次退出程序", Toast.LENGTH_SHORT).show();
            ScheduledExecutorService service = Executors.newScheduledThreadPool(1);
            service.schedule(new Runnable() {
                @Override
                public void run() {
                    if (!isFinishing()) {
                        isExit = false;
                    }
                }
            }, 2000, TimeUnit.MILLISECONDS);
        } else {
            isExit = true;
            this.finish();
        }
    }
    return false;
}

@Test
public void testDelay() throws InterruptedException {
    ScheduledExecutorService ses =
        Executors.newSingleThreadScheduledExecutor();

    SingletonTask st1 = new SingletonTask(ses, new Runnable() {
        @Override
        public void run() {
            ran += 1;
            time = System.nanoTime();
        }
    });
    st1.reschedule(10, TimeUnit.MILLISECONDS);
    assertFalse("Check that task hasn't run yet", ran > 0);

    ses.shutdown();
    ses.awaitTermination(5, TimeUnit.SECONDS);

    assertEquals("Check that task ran", 1, ran);
}

public ExperimentalBitmapAnimationDrawableFactory(
    AnimatedDrawableBackendProvider animatedDrawableBackendProvider,
    ScheduledExecutorService scheduledExecutorServiceForUiThread,
    ExecutorService executorServiceForFramePreparing,
    MonotonicClock monotonicClock,
    PlatformBitmapFactory platformBitmapFactory,
    CountingMemoryCache<CacheKey, CloseableImage> backingCache,
    Supplier<Integer> cachingStrategySupplier,
    Supplier<Integer> numberOfFramesToPrepareSupplier) {
  mAnimatedDrawableBackendProvider = animatedDrawableBackendProvider;
  mScheduledExecutorServiceForUiThread = scheduledExecutorServiceForUiThread;
  mExecutorServiceForFramePreparing = executorServiceForFramePreparing;
  mMonotonicClock = monotonicClock;
  mPlatformBitmapFactory = platformBitmapFactory;
  mBackingCache = backingCache;
  mCachingStrategySupplier = cachingStrategySupplier;
  mNumberOfFramesToPrepareSupplier = numberOfFramesToPrepareSupplier;
}

/**
 * Package private constructor for unit test.
 */
AnomalyDetector(LinkedBlockingDeque<Anomaly> anomalies,
                long anomalyDetectionIntervalMs,
                KafkaCruiseControl kafkaCruiseControl,
                AnomalyNotifier anomalyNotifier,
                GoalViolationDetector goalViolationDetector,
                BrokerFailureDetector brokerFailureDetector,
                ScheduledExecutorService detectorScheduler) {
  _anomalies = anomalies;
  _anomalyDetectionIntervalMs = anomalyDetectionIntervalMs;
  _anomalyNotifier = anomalyNotifier;
  _goalViolationDetector = goalViolationDetector;
  _brokerFailureDetector = brokerFailureDetector;
  _kafkaCruiseControl = kafkaCruiseControl;
  _detectorScheduler = detectorScheduler;
  _shutdown = false;
  _brokerFailureRate = new Meter();
  _goalViolationRate = new Meter();
}

/**
 * Starts notification pull.
 */
public void startNotificationPull() {
    if (isPullingActive()) {
        api.getLogger().logInfo("Notification pull is already working.");
        return;
    }
    final Runnable cachingSingleAction = createCachingSingleAction();
    pullHandle = null;
    if (pullThreads instanceof ScheduledExecutorService) {
        pullHandle = ((ScheduledExecutorService) pullThreads).scheduleWithFixedDelay(cachingSingleAction, 0, 50,
                TimeUnit.MILLISECONDS);
    } else {
        pullHandle = pullThreads.submit(new Runnable() {

            @Override
            public void run() {
                while (true) {
                    cachingSingleAction.run();
                }

            }
        });
    }

}

/**
 * Creates a {@link ScheduledExecutorService} that renames the {@link Thread threads} that its
 * tasks run in.
 *
 * <p>The names are retrieved from the {@code nameSupplier} on the thread that is being renamed
 * right before each task is run. The renaming is best effort, if a {@link SecurityManager}
 * prevents the renaming then it will be skipped but the tasks will still execute.
 *
 *
 * @param service The executor to decorate
 * @param nameSupplier The source of names for each task
 */
@GwtIncompatible // concurrency
static ScheduledExecutorService renamingDecorator(
    final ScheduledExecutorService service, final Supplier<String> nameSupplier) {
  checkNotNull(service);
  checkNotNull(nameSupplier);
  if (isAppEngine()) {
    // AppEngine doesn't support thread renaming, so don't even try.
    return service;
  }
  return new WrappingScheduledExecutorService(service) {
    @Override
    protected <T> Callable<T> wrapTask(Callable<T> callable) {
      return Callables.threadRenaming(callable, nameSupplier);
    }

    @Override
    protected Runnable wrapTask(Runnable command) {
      return Callables.threadRenaming(command, nameSupplier);
    }
  };
}

private <T> void scheduleRetry(
    Throwable e, ScheduledExecutorService retryExecutor,
    CheckedSupplier<? extends CompletionStage<T>, ?> supplier, CompletableFuture<T> future) {
  try {
    Maybe<ExceptionPlan.Execution<Delay<?>>, ?> maybeRetry = plan.execute(e);
    maybeRetry.ifPresent(execution -> {
      future.exceptionally(x -> {
        addSuppressedTo(x, e);
        return null;
      });
      if (future.isDone()) return;  // like, canceled immediately before scheduling.
      @SuppressWarnings("unchecked")  // delay came from upon(), which enforces <? super E>.
      Delay<Throwable> delay = (Delay<Throwable>) execution.strategy();
      Retryer nextRound = new Retryer(execution.remainingExceptionPlan());
      Failable retry = () -> nextRound.invokeWithRetry(supplier, retryExecutor, future);
      delay.asynchronously(e, retry, retryExecutor, future);
    });
    maybeRetry.catching(future::completeExceptionally);
  } catch (Throwable unexpected) {
    addSuppressedTo(unexpected, e);
    throw unexpected;
  }
}

@Override
protected ScheduledExecutorService scheduler() {
  if (this.scheduler == null) {
    synchronized (this) {
      if (this.scheduler == null) {
        ThreadFactory timerFactory = new ThreadFactoryBuilder()
            .setNameFormat("AsyncReporter-" + id + "-timer-%d")
            .setDaemon(true)
            .build();
        ScheduledThreadPoolExecutor timerPool = new ScheduledThreadPoolExecutor(timerThreads, timerFactory);
        timerPool.setRemoveOnCancelPolicy(true);
        this.scheduler = timerPool;
        return timerPool;
      }
    }
  }
  return scheduler;
}

public static ModbusMaster create ( final BundleContext context, final ScheduledExecutorService executor, final String id, final NioProcessor processor, final Map<String, String> parameters ) throws Exception
{
    final ModbusMaster device = new ModbusMaster ( context, id, executor, processor, "ModbusMaster", "modbus" );

    try
    {
        device.configure ( parameters );
    }
    catch ( final Exception e )
    {
        // dispose what was already created
        device.dispose ();
        throw e;
    }

    return device;
}

@Inject
Poller(@IoExecutor Executor ioExecutor,
        @Scheduler ScheduledExecutorService scheduler,
        ConnectionManager connectionManager,
        ConnectionRegistry connectionRegistry, PluginManager pluginManager,
        SecureRandom random, Clock clock) {
    this.ioExecutor = ioExecutor;
    this.scheduler = scheduler;
    this.connectionManager = connectionManager;
    this.connectionRegistry = connectionRegistry;
    this.pluginManager = pluginManager;
    this.random = random;
    this.clock = clock;
    lock = new ReentrantLock();
    tasks = new HashMap<TransportId, PollTask>();
}

protected MackerelReporter(MetricRegistry registry,
                           MackerelSender mackerel,
                           Clock clock,
                           String prefix,
                           TimeUnit rateUnit,
                           TimeUnit durationUnit,
                           MetricFilter filter,
                           ScheduledExecutorService executor,
                           boolean shutdownExecutorOnStop,
                           Set<MetricAttribute> disabledMetricAttributes) {
    super(registry, "mackerel-reporter", filter, rateUnit, durationUnit, executor, shutdownExecutorOnStop,
            disabledMetricAttributes);
    this.mackerel = mackerel;
    this.clock = clock;
    this.prefix = prefix;
}

@Test
public void testBasic() throws InterruptedException {
    ScheduledExecutorService ses =
        Executors.newSingleThreadScheduledExecutor();

    SingletonTask st1 = new SingletonTask(ses, new Runnable() {
        @Override
        public void run() {
            ran += 1;
        }
    });
    st1.reschedule(0, null);
    ses.shutdown();
    ses.awaitTermination(5, TimeUnit.SECONDS);

    assertEquals("Check that task ran", 1, ran);
}

public JdbcQuery ( final JdbcDao jdbcStorageDao, final Filter filter, final ScheduledExecutorService executor, final List<JdbcQuery> openQueries ) throws SQLException, NotSupportedException
{
    openQueries.add ( this );
    this.openQueries = new WeakReference<List<JdbcQuery>> ( openQueries );
    this.resultSet = jdbcStorageDao.queryEvents ( filter );
    this.statement = this.resultSet.getStatement ();
    this.hasMore = this.resultSet.next ();
    this.future = executor.schedule ( new Callable<Boolean> () {
        @Override
        public Boolean call ()
        {
            logger.warn ( "Query '{}' was open for over an hour, or service is being shut down, and will now be closed automatically" );
            dispose ();
            return true;
        }
    }, 1, TimeUnit.HOURS );
}

/**
 * Test of read method, of class UART.
 *
 * @throws java.lang.InterruptedException
 * @throws java.util.concurrent.ExecutionException
 * @throws java.util.concurrent.TimeoutException
 */
@Test
public void testRead() throws InterruptedException, ExecutionException, TimeoutException {
    ExecutorService executor = Executors.newSingleThreadExecutor();
    Semaphore semaphore = new Semaphore(0);
    mockInput = '5';
    Future<Void> future = executor.submit(() -> {
        ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
        ScheduledFuture<?> releaserHandle = scheduler.schedule(() -> {
            semaphore.release();
        }, 5, TimeUnit.SECONDS);
        uart.read();
        releaserHandle.get(5, TimeUnit.SECONDS);
        return null;
    });
    semaphore.tryAcquire(15, TimeUnit.SECONDS);
    uart.feed(mockInput);
    future.get(25, TimeUnit.SECONDS);
    assertEquals("UART read result is wrong.", mockInput, registerFile.getValue("r0"));
}

public void runTest()
{
    ScheduledExecutorService s=new ScheduledThreadPoolExecutor(2, new NamedThreadFactory("Scheduled"));
    ExecutorService es=Executors.newCachedThreadPool();
    es.execute(this::writeTest);
    es.execute(this::writeTest);
    es.execute(this::writeTest);
    es.execute(this::writeTest);
    es.execute(this::readTest);
    es.execute(this::readTest);
    es.execute(this::readTest);
    es.execute(this::readTest);
    es.execute(this::readTest);
    es.execute(this::readTest);
    s.scheduleAtFixedRate(this::printInfo,5, 5, TimeUnit.SECONDS);
}

public ScriptSourceFactory ( final BundleContext context, final ScheduledExecutorService executor, final EventProcessor eventProcessor ) throws InvalidSyntaxException
{
    super ( context );
    this.executor = executor;

    this.eventProcessor = eventProcessor;

    this.objectPool = new ObjectPoolImpl<DataSource> ();
    this.poolRegistration = ObjectPoolHelper.registerObjectPool ( context, this.objectPool, DataSource.class );

    this.poolTracker = new ObjectPoolTracker<DataSource> ( context, DataSource.class.getName () );
    this.poolTracker.open ();
}

public HSDBItemController ( final String id, final ScheduledExecutorService executor, final BundleContext context, final HSDBValueSource source )
{
    this.source = source;

    final Map<String, Variant> properties = new HashMap<String, Variant> ();

    final HistoricalItemInformation information = new HistoricalItemInformation ( id, properties );
    this.item = new HSDBHistoricalItem ( executor, source, information );

    final Dictionary<String, Object> serviceProperties = new Hashtable<String, Object> ();
    serviceProperties.put ( Constants.SERVICE_PID, id );
    serviceProperties.put ( Constants.SERVICE_VENDOR, "Eclipse SCADA Project" );
    this.handle = context.registerService ( HistoricalItem.class, this.item, serviceProperties );
}

private StaticModbusExport ( final ScheduledExecutorService executor, final IoProcessor<NioSession> processor, final HiveSource hiveSource, final ObjectPoolDataItemFactory itemFactory, final boolean disposeProcessor )
{
    super ( executor, processor, hiveSource, itemFactory );
    this.executor = executor;
    this.processor = processor;
    this.disposeProcessor = disposeProcessor;
}

@Override
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable task, long delay) {
    ScheduledExecutorService executor = getScheduledExecutor();
    try {
        return executor.scheduleWithFixedDelay(errorHandlingTask(task, true), 0, delay, TimeUnit.MILLISECONDS);
    }
    catch (RejectedExecutionException ex) {
        throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex);
    }
}

@Test
public void testTimeout()
        throws Exception
{
    ScheduledExecutorService executor = newSingleThreadScheduledExecutor(daemonThreadsNamed("test-timeout"));

    DriftNettyMethodInvoker invoker = new DriftNettyMethodInvoker(
            new HangingConnectionManager(),
            executor,
            new Duration(20, MILLISECONDS));

    ListenableFuture<Object> response = invoker.invoke(new InvokeRequest(
            new MethodMetadata(
                    "test",
                    ImmutableList.of(),
                    (ThriftCodec<Object>) (Object) new VoidThriftCodec(),
                    ImmutableMap.of(),
                    false),
            () -> HostAndPort.fromParts("localhost", 1234),
            ImmutableMap.of(),
            ImmutableList.of()));

    try {
        response.get();
        fail("expected exception");
    }
    catch (ExecutionException e) {
        assertInstanceOf(e.getCause(), io.airlift.drift.TException.class);
        assertEquals(e.getCause().getMessage(), "Invocation response future did not complete after 20.00ms");
    }
    finally {
        executor.shutdown();
    }
}

public static <T extends AnimationBackend>
AnimationBackendDelegate<T> createForBackend(
    T backend,
    InactivityListener inactivityListener,
    MonotonicClock monotonicClock,
    ScheduledExecutorService scheduledExecutorServiceForUiThread) {
  return new AnimationBackendDelegateWithInactivityCheck<>(
      backend,
      inactivityListener,
      monotonicClock,
      scheduledExecutorServiceForUiThread);
}

@Test(expected = RuntimeException.class)
public void cancellation() throws CheckedFutureException {
  final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
  try {
    final Future<Long> f = ds.execute("SELECT pg_sleep(999)");
    f.raise(new RuntimeException());
    f.get(timeout);
  } finally {
    scheduler.shutdown();
  }
}

public static void main(String[] args) {
    final RecorderSchedule s = new RecorderSchedule();
    ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("recoder"));
    scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
        public void run() {
            try {
                s.getDelaySecond();
            } catch (Exception e) {
                logger.error("RecoderSchedule error", e);
            }
        }
    }, 0, 1, TimeUnit.SECONDS);
}

public static void registerExecutor(ScheduledThreadPoolExecutor service) {
    while (((ScheduledExecutorService) PURGE.get()) == null) {
        ScheduledExecutorService exec = Executors.newScheduledThreadPool(1, new RxThreadFactory(PURGE_THREAD_PREFIX));
        if (PURGE.compareAndSet(null, exec)) {
            exec.scheduleAtFixedRate(new Runnable() {
                public void run() {
                    NewThreadWorker.purgeExecutors();
                }
            }, (long) PURGE_FREQUENCY, (long) PURGE_FREQUENCY, TimeUnit.MILLISECONDS);
            break;
        }
        exec.shutdownNow();
    }
    EXECUTORS.putIfAbsent(service, service);
}

@Test
public void readPositionEnd() throws IOException, InterruptedException {
    final Utils.LogFile logFile = new Utils.LogFile(100 * 1024, 400, 100);
    logFile.close();
    final ArrayBlockingQueue<FileChunk> chunkQueue = Queues.newArrayBlockingQueue(1);
    final AsynchronousFileChannel channel = AsynchronousFileChannel.open(logFile.getPath(), StandardOpenOption.READ);
    final CountingAsyncFileChannel spy = new CountingAsyncFileChannel(channel);

    final ChunkReader chunkReader = new ChunkReader(mock(FileInput.class), logFile.getPath(), spy, chunkQueue, 10 * 1024,
            FileInput.InitialReadPosition.END, null);

    final ScheduledExecutorService chunkReaderExecutor = Executors.newSingleThreadScheduledExecutor(
            new ThreadFactoryBuilder()
                    .setDaemon(false)
                    .setNameFormat("file-chunk-reader-%d")
                    .setUncaughtExceptionHandler(this)
                    .build()
    );

    final Thread consumer = new Thread() {
        @Override
        public void run() {
            try {
                final FileChunk chunk = chunkQueue.poll(2, TimeUnit.SECONDS);
                assertNull("Reading from the end of the file must not produce a chunk for a non-changing file.", chunk);
            } catch (InterruptedException ignore) {
            }
        }
    };
    consumer.start();
    chunkReaderExecutor.scheduleAtFixedRate(chunkReader, 0, 250, TimeUnit.MILLISECONDS);
    consumer.join();

    // we can process one chunk at a time, so one read is queued, the second is buffered
    assertEquals("The e should be empty", 1, chunkQueue.remainingCapacity());
}

static void deployContinuously(Ignite ignite) {
    long reportFrequency = TestsHelper.getLoadTestsStatisticsReportFrequency();
    ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);
    final StatisticsDeploymentHelper helper = new StatisticsDeploymentHelper(ignite);

    executor.scheduleAtFixedRate(new Runnable() {
        @Override public void run() {
            helper.deployStatisticsCollector();
        }
    }, 0, reportFrequency, TimeUnit.MILLISECONDS);
}

/**
 * Trial run of the engine. Shuts down after {@code timeout} seconds after startup.
 *
 * @param engine the engine.
 * @param timeout timeout in seconds.
 */
public static void trialRunEngine(Engine engine, int timeout) {
    final Semaphore semaphore = new Semaphore(0, true);

    // Startup the engine. After startup the engine runs on the threads other than the current one.
    engine.startup();

    try {
        ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
        executor.schedule(() -> {
            // Release the semaphore after timeout.
            semaphore.release();
        }, timeout, TimeUnit.SECONDS);

        try {
            // Wait for releasing the semaphore after timeout.
            semaphore.acquire();
        } catch (InterruptedException e) {
            logger.warn("trialRunEngine", e);
        }

        executor.shutdown();
    } finally {
        // Shutdown the engine.
        engine.shutdown();
    }
}

@VisibleForTesting
FactsObserverFactory(@NonNull LinkFactory<FactsResource> factsResourceLinkFactory,
        @NonNull HyperSchemaCreator hyperSchemaCreator,
        @NonNull FactTransformer factTransformer,
        @NonNull ScheduledExecutorService executorService, int waitSecondsForCleanUpCheck) {
    this.factsResourceLinkFactory = factsResourceLinkFactory;
    this.hyperSchemaCreator = hyperSchemaCreator;
    this.factTransformer = factTransformer;
    this.executorService = executorService;
    this.waitSecondsForCleanUpCheck = waitSecondsForCleanUpCheck;
}

public HSDBStorageManager ( final ScheduledExecutorService executor, final String prefix, final BundleContext bundleContext, final File root )
{
    this.executor = executor;
    this.prefix = prefix;
    this.root = root;
    this.context = bundleContext;

    scan ();
}

final void asynchronously(
    E event, Failable retry, ScheduledExecutorService executor, CompletableFuture<?> result) {
  beforeDelay(event);
  Failable afterDelay = () -> {
    afterDelay(event);
    retry.run();
  };
  ScheduledFuture<?> scheduled = executor.schedule(
      () -> afterDelay.run(result::completeExceptionally),
      duration().toMillis(), TimeUnit.MILLISECONDS);
  ifCancelled(result, canceled -> {scheduled.cancel(true);});
}

@Override
final Future<?> schedule(
    AbstractService service, ScheduledExecutorService executor, Runnable runnable) {
  ReschedulableCallable task = new ReschedulableCallable(service, executor, runnable);
  task.reschedule();
  return task;
}

@Override
public void afterPropertiesSet() throws NamingException {
    if (this.jndiName != null) {
        ScheduledExecutorService executor = this.jndiLocator.lookup(this.jndiName, ScheduledExecutorService.class);
        setConcurrentExecutor(executor);
        setScheduledExecutor(executor);
    }
}

public static void main(String[] args) {
    ScheduledExecutorService ses= Executors.newScheduledThreadPool(10);
    ses.scheduleWithFixedDelay(new Runnable() {
        public void run() {
            try {
                Thread.sleep(1000);
                System.out.println(System.currentTimeMillis()/1000);
            }catch (InterruptedException e){
                e.printStackTrace();
            }
        }
    },0,2, TimeUnit.SECONDS);
}

@Test
public void testPollFromAny() throws InterruptedException {
    final RBoundedBlockingQueue<Integer> queue1 = redisson.getBoundedBlockingQueue("queue:pollany");
    assertThat(queue1.trySetCapacity(10)).isTrue();
    ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
    executor.schedule(() -> {
        RBoundedBlockingQueue<Integer> queue2 = redisson.getBoundedBlockingQueue("queue:pollany1");
        assertThat(queue2.trySetCapacity(10)).isTrue();
        RBoundedBlockingQueue<Integer> queue3 = redisson.getBoundedBlockingQueue("queue:pollany2");
        assertThat(queue3.trySetCapacity(10)).isTrue();
        try {
            queue3.put(2);
            queue1.put(1);
            queue2.put(3);
        } catch (Exception e) {
            Assert.fail();
        }
    }, 3, TimeUnit.SECONDS);

    long s = System.currentTimeMillis();
    int l = queue1.pollFromAny(40, TimeUnit.SECONDS, "queue:pollany1", "queue:pollany2");

    Assert.assertEquals(2, l);
    Assert.assertTrue(System.currentTimeMillis() - s > 2000);

    executor.shutdown();
    assertThat(executor.awaitTermination(1, TimeUnit.MINUTES)).isTrue();
}

@VisibleForTesting
public static InstanceManager create(
    AthenaXConfiguration conf,
    InstanceStateUpdateListener listener,
    ScheduledExecutorService executor) {
  HashMap<String, ClusterInfo> c = new HashMap<>();
  for (Map.Entry<String, AthenaXConfiguration.YarnCluster> e : conf.clusters().entrySet()) {
    ClusterInfo ci = new ClusterInfo(e.getKey(), e.getValue().toYarnClusterConfiguration());
    c.put(e.getKey(), ci);
  }
  return new InstanceManager(c, listener, executor, conf.getExtraConfLong(INSTANCE_MANAGER_RESCAN_INTERVAL));
}

JobDeployer(YarnClusterConfiguration clusterConf, YarnClient yarnClient,
            ScheduledExecutorService executor, Configuration flinkConf) {
  this.clusterConf = clusterConf;
  this.executor = executor;
  this.flinkConf = flinkConf;
  this.yarnClient = yarnClient;
}

public BufferedDataSourceImpl ( final BundleContext context, final ScheduledExecutorService scheduler, final ObjectPoolTracker<DataSource> poolTracker, final DataNodeTracker dataNodeTracker, final String configurationId, final ObjectPoolImpl<BufferedDataSource> objectPool )
{
    this.context = context;
    this.scheduler = scheduler;
    this.poolTracker = poolTracker;
    this.dataNodeTracker = dataNodeTracker;
    this.configurationId = configurationId;
    this.objectPool = objectPool;
}