@Override
    public void schedulePendingRequestBulkTimeoutCheck(
            final PhysicalSlotRequestBulk bulk, Time timeout) {
        PhysicalSlotRequestBulkWithTimestamp bulkWithTimestamp =
                new PhysicalSlotRequestBulkWithTimestamp(bulk);
        bulkWithTimestamp.markUnfulfillable(clock.relativeTimeMillis());
        schedulePendingRequestBulkWithTimestampCheck(bulkWithTimestamp, timeout);
    }
@Test
    void testUnfulfillableBulkIsCancelled() {
        final CompletableFuture<SlotRequestId> cancellationFuture = new CompletableFuture<>();
        final SlotRequestId slotRequestId = new SlotRequestId();
        final PhysicalSlotRequestBulk bulk =
                createPhysicalSlotRequestBulkWithCancellationFuture(
                        cancellationFuture, slotRequestId);
        bulkChecker.schedulePendingRequestBulkTimeoutCheck(bulk, TIMEOUT);
        clock.advanceTime(TIMEOUT.toMilliseconds() + 1L, TimeUnit.MILLISECONDS);
        assertThat(cancellationFuture.join()).isEqualTo(slotRequestId);
    }