@Override
    public void schedulePendingRequestBulkTimeoutCheck(
            final PhysicalSlotRequestBulk bulk, Time timeout) {
        PhysicalSlotRequestBulkWithTimestamp bulkWithTimestamp =
                new PhysicalSlotRequestBulkWithTimestamp(bulk);
        bulkWithTimestamp.markUnfulfillable(clock.relativeTimeMillis());
        schedulePendingRequestBulkWithTimestampCheck(bulkWithTimestamp, timeout);
    }
@Test
    void testFulfilledBulkIsNotCancelled() throws InterruptedException, ExecutionException {
        final CompletableFuture<SlotRequestId> cancellationFuture = new CompletableFuture<>();
        final PhysicalSlotRequestBulk bulk =
                createPhysicalSlotRequestBulkWithCancellationFuture(
                        cancellationFuture, new SlotRequestId());
        bulkChecker.schedulePendingRequestBulkTimeoutCheck(bulk, TIMEOUT);
        checkNotCancelledAfter(cancellationFuture, 2 * TIMEOUT.toMilliseconds());
    }