@Test public void testCheckIdleSlot() throws Exception { final ManualClock clock = new ManualClock(); try (TestingSlotPoolImpl slotPool = createSlotPoolImpl(clock)) { final BlockingQueue<AllocationID> freedSlots = new ArrayBlockingQueue<>(1); taskManagerGateway.setFreeSlotFunction( (AllocationID allocationId, Throwable cause) -> { try { freedSlots.put(allocationId); return CompletableFuture.completedFuture(Acknowledge.get()); } catch (InterruptedException e) { return FutureUtils.completedExceptionally(e); } }); setupSlotPool(slotPool, resourceManagerGateway, mainThreadExecutor); final AllocationID expiredSlotID = new AllocationID(); final AllocationID freshSlotID = new AllocationID(); final SlotOffer slotToExpire = new SlotOffer(expiredSlotID, 0, ResourceProfile.UNKNOWN); final SlotOffer slotToNotExpire = new SlotOffer(freshSlotID, 1, ResourceProfile.UNKNOWN); assertThat(slotPool.registerTaskManager(taskManagerLocation.getResourceID()), Matchers.is(true)); assertThat( slotPool.offerSlot(taskManagerLocation, taskManagerGateway, slotToExpire), Matchers.is(true)); clock.advanceTime(timeout.toMilliseconds(), TimeUnit.MILLISECONDS); assertThat(slotPool.offerSlot(taskManagerLocation, taskManagerGateway, slotToNotExpire), Matchers.is(true)); clock.advanceTime(1L, TimeUnit.MILLISECONDS); slotPool.triggerCheckIdleSlot(); final AllocationID freedSlot = freedSlots.poll(timeout.toMilliseconds(), TimeUnit.MILLISECONDS); assertThat(freedSlot, Matchers.is(expiredSlotID)); assertThat(freedSlots.isEmpty(), Matchers.is(true)); } }
protected void checkIdleSlot() { final long currentRelativeTimeMillis = clock.relativeTimeMillis(); final List<AllocatedSlot> expiredSlots = new ArrayList<>(availableSlots.size()); for (SlotAndTimestamp slotAndTimestamp : availableSlots.availableSlots.values()) { if (currentRelativeTimeMillis - slotAndTimestamp.timestamp > idleSlotTimeout.toMilliseconds()) { expiredSlots.add(slotAndTimestamp.slot); } } final FlinkException cause = new FlinkException("Releasing idle slot."); for (AllocatedSlot expiredSlot : expiredSlots) { final AllocationID allocationID = expiredSlot.getAllocationId(); if (availableSlots.tryRemove(allocationID) != null) { log.info("Releasing idle slot [{}].", allocationID); final CompletableFuture<Acknowledge> freeSlotFuture = expiredSlot.getTaskManagerGateway().freeSlot( allocationID, cause, rpcTimeout); FutureUtils.whenCompleteAsyncIfNotDone( freeSlotFuture, componentMainThreadExecutor, (Acknowledge ignored, Throwable throwable) -> { if (throwable != null) { log.debug("Releasing slot [{}] of registered TaskExecutor {} failed. Discarding slot.", allocationID, expiredSlot.getTaskManagerId(), throwable); } }); } } scheduleRunAsync(this::checkIdleSlot, idleSlotTimeout); }
SlotPoolImpl implements SlotPool { protected void checkIdleSlot() { final long currentRelativeTimeMillis = clock.relativeTimeMillis(); final List<AllocatedSlot> expiredSlots = new ArrayList<>(availableSlots.size()); for (SlotAndTimestamp slotAndTimestamp : availableSlots.availableSlots.values()) { if (currentRelativeTimeMillis - slotAndTimestamp.timestamp > idleSlotTimeout.toMilliseconds()) { expiredSlots.add(slotAndTimestamp.slot); } } final FlinkException cause = new FlinkException("Releasing idle slot."); for (AllocatedSlot expiredSlot : expiredSlots) { final AllocationID allocationID = expiredSlot.getAllocationId(); if (availableSlots.tryRemove(allocationID) != null) { log.info("Releasing idle slot [{}].", allocationID); final CompletableFuture<Acknowledge> freeSlotFuture = expiredSlot.getTaskManagerGateway().freeSlot( allocationID, cause, rpcTimeout); FutureUtils.whenCompleteAsyncIfNotDone( freeSlotFuture, componentMainThreadExecutor, (Acknowledge ignored, Throwable throwable) -> { if (throwable != null) { log.debug("Releasing slot [{}] of registered TaskExecutor {} failed. Discarding slot.", allocationID, expiredSlot.getTaskManagerId(), throwable); } }); } } scheduleRunAsync(this::checkIdleSlot, idleSlotTimeout); } }
SlotPoolImpl implements SlotPool { protected void checkIdleSlot() { final long currentRelativeTimeMillis = clock.relativeTimeMillis(); final List<AllocatedSlot> expiredSlots = new ArrayList<>(availableSlots.size()); for (SlotAndTimestamp slotAndTimestamp : availableSlots.availableSlots.values()) { if (currentRelativeTimeMillis - slotAndTimestamp.timestamp > idleSlotTimeout.toMilliseconds()) { expiredSlots.add(slotAndTimestamp.slot); } } final FlinkException cause = new FlinkException("Releasing idle slot."); for (AllocatedSlot expiredSlot : expiredSlots) { final AllocationID allocationID = expiredSlot.getAllocationId(); if (availableSlots.tryRemove(allocationID) != null) { log.info("Releasing idle slot [{}].", allocationID); final CompletableFuture<Acknowledge> freeSlotFuture = expiredSlot.getTaskManagerGateway().freeSlot( allocationID, cause, rpcTimeout); FutureUtils.whenCompleteAsyncIfNotDone( freeSlotFuture, componentMainThreadExecutor, (Acknowledge ignored, Throwable throwable) -> { if (throwable != null) { log.debug("Releasing slot [{}] of registered TaskExecutor {} failed. Discarding slot.", allocationID, expiredSlot.getTaskManagerId(), throwable); } }); } } scheduleRunAsync(this::checkIdleSlot, idleSlotTimeout); } SlotPoolImpl(
			JobID jobId,
			Clock clock,
			Time rpcTimeout,
			Time idleSlotTimeout,
			Time batchSlotTimeout); }
SlotPoolImpl implements SlotPool { protected void checkIdleSlot() { final long currentRelativeTimeMillis = clock.relativeTimeMillis(); final List<AllocatedSlot> expiredSlots = new ArrayList<>(availableSlots.size()); for (SlotAndTimestamp slotAndTimestamp : availableSlots.availableSlots.values()) { if (currentRelativeTimeMillis - slotAndTimestamp.timestamp > idleSlotTimeout.toMilliseconds()) { expiredSlots.add(slotAndTimestamp.slot); } } final FlinkException cause = new FlinkException("Releasing idle slot."); for (AllocatedSlot expiredSlot : expiredSlots) { final AllocationID allocationID = expiredSlot.getAllocationId(); if (availableSlots.tryRemove(allocationID) != null) { log.info("Releasing idle slot [{}].", allocationID); final CompletableFuture<Acknowledge> freeSlotFuture = expiredSlot.getTaskManagerGateway().freeSlot( allocationID, cause, rpcTimeout); FutureUtils.whenCompleteAsyncIfNotDone( freeSlotFuture, componentMainThreadExecutor, (Acknowledge ignored, Throwable throwable) -> { if (throwable != null) { log.debug("Releasing slot [{}] of registered TaskExecutor {} failed. Discarding slot.", allocationID, expiredSlot.getTaskManagerId(), throwable); } }); } } scheduleRunAsync(this::checkIdleSlot, idleSlotTimeout); } SlotPoolImpl(
			JobID jobId,
			Clock clock,
			Time rpcTimeout,
			Time idleSlotTimeout,
			Time batchSlotTimeout); void start(
		@Nonnull JobMasterId jobMasterId,
		@Nonnull String newJobManagerAddress,
		@Nonnull ComponentMainThreadExecutor componentMainThreadExecutor); @Override void suspend(); @Override void close(); @Override void connectToResourceManager(@Nonnull ResourceManagerGateway resourceManagerGateway); @Override void disconnectResourceManager(); @Override void releaseSlot(@Nonnull SlotRequestId slotRequestId, @Nullable Throwable cause); @Override Optional<PhysicalSlot> allocateAvailableSlot(
		@Nonnull SlotRequestId slotRequestId,
		@Nonnull AllocationID allocationID); @Nonnull @Override CompletableFuture<PhysicalSlot> requestNewAllocatedSlot(
			@Nonnull SlotRequestId slotRequestId,
			@Nonnull ResourceProfile resourceProfile,
			Time timeout); @Nonnull @Override CompletableFuture<PhysicalSlot> requestNewAllocatedBatchSlot(
		@Nonnull SlotRequestId slotRequestId,
		@Nonnull ResourceProfile resourceProfile); @Override @Nonnull Collection<SlotInfo> getAvailableSlotsInformation(); @Override Collection<SlotOffer> offerSlots(
			TaskManagerLocation taskManagerLocation,
			TaskManagerGateway taskManagerGateway,
			Collection<SlotOffer> offers); @Override Optional<ResourceID> failAllocation(final AllocationID allocationID, final Exception cause); @Override boolean registerTaskManager(final ResourceID resourceID); @Override boolean releaseTaskManager(final ResourceID resourceId, final Exception cause); @Override AllocatedSlotReport createAllocatedSlotReport(ResourceID taskManagerId); }
SlotPoolImpl implements SlotPool { protected void checkIdleSlot() { final long currentRelativeTimeMillis = clock.relativeTimeMillis(); final List<AllocatedSlot> expiredSlots = new ArrayList<>(availableSlots.size()); for (SlotAndTimestamp slotAndTimestamp : availableSlots.availableSlots.values()) { if (currentRelativeTimeMillis - slotAndTimestamp.timestamp > idleSlotTimeout.toMilliseconds()) { expiredSlots.add(slotAndTimestamp.slot); } } final FlinkException cause = new FlinkException("Releasing idle slot."); for (AllocatedSlot expiredSlot : expiredSlots) { final AllocationID allocationID = expiredSlot.getAllocationId(); if (availableSlots.tryRemove(allocationID) != null) { log.info("Releasing idle slot [{}].", allocationID); final CompletableFuture<Acknowledge> freeSlotFuture = expiredSlot.getTaskManagerGateway().freeSlot( allocationID, cause, rpcTimeout); FutureUtils.whenCompleteAsyncIfNotDone( freeSlotFuture, componentMainThreadExecutor, (Acknowledge ignored, Throwable throwable) -> { if (throwable != null) { log.debug("Releasing slot [{}] of registered TaskExecutor {} failed. Discarding slot.", allocationID, expiredSlot.getTaskManagerId(), throwable); } }); } } scheduleRunAsync(this::checkIdleSlot, idleSlotTimeout); } SlotPoolImpl(
			JobID jobId,
			Clock clock,
			Time rpcTimeout,
			Time idleSlotTimeout,
			Time batchSlotTimeout); void start(
		@Nonnull JobMasterId jobMasterId,
		@Nonnull String newJobManagerAddress,
		@Nonnull ComponentMainThreadExecutor componentMainThreadExecutor); @Override void suspend(); @Override void close(); @Override void connectToResourceManager(@Nonnull ResourceManagerGateway resourceManagerGateway); @Override void disconnectResourceManager(); @Override void releaseSlot(@Nonnull SlotRequestId slotRequestId, @Nullable Throwable cause); @Override Optional<PhysicalSlot> allocateAvailableSlot(
		@Nonnull SlotRequestId slotRequestId,
		@Nonnull AllocationID allocationID); @Nonnull @Override CompletableFuture<PhysicalSlot> requestNewAllocatedSlot(
			@Nonnull SlotRequestId slotRequestId,
			@Nonnull ResourceProfile resourceProfile,
			Time timeout); @Nonnull @Override CompletableFuture<PhysicalSlot> requestNewAllocatedBatchSlot(
		@Nonnull SlotRequestId slotRequestId,
		@Nonnull ResourceProfile resourceProfile); @Override @Nonnull Collection<SlotInfo> getAvailableSlotsInformation(); @Override Collection<SlotOffer> offerSlots(
			TaskManagerLocation taskManagerLocation,
			TaskManagerGateway taskManagerGateway,
			Collection<SlotOffer> offers); @Override Optional<ResourceID> failAllocation(final AllocationID allocationID, final Exception cause); @Override boolean registerTaskManager(final ResourceID resourceID); @Override boolean releaseTaskManager(final ResourceID resourceId, final Exception cause); @Override AllocatedSlotReport createAllocatedSlotReport(ResourceID taskManagerId);  }