@Test void shouldNotRecordOnEmptyPool() { CompletableFuture<Void> completableFuture = new CompletableFuture<>(); completableFuture.complete(null); assertThrows( NonRetryableException.class, () -> workerPool.record(mock(Event.class), (w, e) -> completableFuture)); }
CompletableFuture<Void> record( Event event, BiFunction<MantisWorker, Event, CompletableFuture<Void>> function) throws NonRetryableException { MantisWorker worker = getRandomWorker(); if (worker == null) { throw new NonRetryableException("no available workers in pool"); } CompletableFuture<Void> future = function.apply(worker, event); future.whenCompleteAsync((v, t) -> { if (t != null) { pool.put(worker, pool.get(worker) + 1); if (shouldBlacklist(worker)) { eventChannel.close(worker); pool.remove(worker); workerPoolGauge.set((double) pool.size()); blacklist.add(worker); blacklistedWorkersGauge.set((double) blacklist.size()); } } }); return future; }
ChoiceOfTwoWorkerPool { CompletableFuture<Void> record( Event event, BiFunction<MantisWorker, Event, CompletableFuture<Void>> function) throws NonRetryableException { MantisWorker worker = getRandomWorker(); if (worker == null) { throw new NonRetryableException("no available workers in pool"); } CompletableFuture<Void> future = function.apply(worker, event); future.whenCompleteAsync((v, t) -> { if (t != null) { pool.put(worker, pool.get(worker) + 1); if (shouldBlacklist(worker)) { eventChannel.close(worker); pool.remove(worker); workerPoolGauge.set((double) pool.size()); blacklist.add(worker); blacklistedWorkersGauge.set((double) blacklist.size()); } } }); return future; } }
ChoiceOfTwoWorkerPool { CompletableFuture<Void> record( Event event, BiFunction<MantisWorker, Event, CompletableFuture<Void>> function) throws NonRetryableException { MantisWorker worker = getRandomWorker(); if (worker == null) { throw new NonRetryableException("no available workers in pool"); } CompletableFuture<Void> future = function.apply(worker, event); future.whenCompleteAsync((v, t) -> { if (t != null) { pool.put(worker, pool.get(worker) + 1); if (shouldBlacklist(worker)) { eventChannel.close(worker); pool.remove(worker); workerPoolGauge.set((double) pool.size()); blacklist.add(worker); blacklistedWorkersGauge.set((double) blacklist.size()); } } }); return future; } ChoiceOfTwoWorkerPool(MrePublishConfiguration config, Registry registry, EventChannel eventChannel); }
ChoiceOfTwoWorkerPool { CompletableFuture<Void> record( Event event, BiFunction<MantisWorker, Event, CompletableFuture<Void>> function) throws NonRetryableException { MantisWorker worker = getRandomWorker(); if (worker == null) { throw new NonRetryableException("no available workers in pool"); } CompletableFuture<Void> future = function.apply(worker, event); future.whenCompleteAsync((v, t) -> { if (t != null) { pool.put(worker, pool.get(worker) + 1); if (shouldBlacklist(worker)) { eventChannel.close(worker); pool.remove(worker); workerPoolGauge.set((double) pool.size()); blacklist.add(worker); blacklistedWorkersGauge.set((double) blacklist.size()); } } }); return future; } ChoiceOfTwoWorkerPool(MrePublishConfiguration config, Registry registry, EventChannel eventChannel);  }
ChoiceOfTwoWorkerPool { CompletableFuture<Void> record( Event event, BiFunction<MantisWorker, Event, CompletableFuture<Void>> function) throws NonRetryableException { MantisWorker worker = getRandomWorker(); if (worker == null) { throw new NonRetryableException("no available workers in pool"); } CompletableFuture<Void> future = function.apply(worker, event); future.whenCompleteAsync((v, t) -> { if (t != null) { pool.put(worker, pool.get(worker) + 1); if (shouldBlacklist(worker)) { eventChannel.close(worker); pool.remove(worker); workerPoolGauge.set((double) pool.size()); blacklist.add(worker); blacklistedWorkersGauge.set((double) blacklist.size()); } } }); return future; } ChoiceOfTwoWorkerPool(MrePublishConfiguration config, Registry registry, EventChannel eventChannel);   }