@Override
    public void cancel(Throwable cause) {
        // pending requests must be canceled first otherwise they might be fulfilled by
        // allocated slots released from this bulk
        for (ExecutionSlotSharingGroup group : pendingRequests.keySet()) {
            for (ExecutionVertexID id : executions.get(group)) {
                logicalSlotRequestCanceller.accept(id, cause);
            }
        }
        for (ExecutionSlotSharingGroup group : fulfilledRequests.keySet()) {
            for (ExecutionVertexID id : executions.get(group)) {
                logicalSlotRequestCanceller.accept(id, cause);
            }
        }
    }
@Test
    void testCancel() {
        LogicalSlotRequestCanceller canceller = new LogicalSlotRequestCanceller();
        SharingPhysicalSlotRequestBulk bulk = createBulk(canceller);
        bulk.markFulfilled(SG1, new AllocationID());
        Throwable cause = new Throwable();
        bulk.cancel(cause);
        assertThat(canceller.cancellations)
                .contains(Tuple2.of(EV1, cause), Tuple2.of(EV2, cause), Tuple2.of(EV4, cause));
    }