AbstractCurrentMetaObserver extends AbstractLifecycleObservable implements Observer, ClusterTypeAware { protected boolean registerJob(String clusterId, String shardId, Releasable releasable) { try { currentMetaManager.addResource(clusterId, shardId, releasable); } catch (Exception e) { try { logger.info("[registerJob][{}][{}] cancel job registration", clusterId, shardId, e); releasable.release(); } catch (Throwable t) { logger.warn("[registerJob][{}][{}]", clusterId, shardId, t); } return false; } return true; }  @SuppressWarnings("rawtypes") @Override void update(Object args, Observable observable); void setCurrentMetaManager(CurrentMetaManager currentMetaManager); @Override int getOrder();  }
@Test public void testRegisterJob() throws Exception { doThrow(new IllegalArgumentException("No such cluster/shard")).when(currentMetaManager) .addResource(anyString(), anyString(), any(Releasable.class)); AtomicInteger innerCounter = new AtomicInteger(); ReleasableCounter counter = new ReleasableCounter(innerCounter); counter.start(); observer.registerJob("cluster", "shard", counter); int waitingTasks = 5; sleep(waitingTasks * 10); Assert.assertTrue(innerCounter.get() < waitingTasks); }