@Override
    public void release(Throwable cause) {
        LOG.debug("Release shared slot ({})", physicalSlotRequestId);
        Preconditions.checkState(
                state == State.ALLOCATED, "The shared slot has already been released.");

        // ensures that we won't call the external release callback if there are still logical slots
        // to release
        state = State.RELEASING;

        // copy the logical slot collection to avoid ConcurrentModificationException
        // if logical slot releases cause cancellation of other executions
        // which will try to call returnLogicalSlot and modify allocatedLogicalSlots collection
        final List<LogicalSlot> logicalSlotsToRelease =
                new ArrayList<>(allocatedLogicalSlots.values());
        for (LogicalSlot allocatedLogicalSlot : logicalSlotsToRelease) {
            // this will also cause the logical slot to be returned
            allocatedLogicalSlot.releaseSlot(cause);
        }
        allocatedLogicalSlots.clear();

        state = State.RELEASED;
    }
@Test
    void testReleaseDoesNotTriggersExternalRelease() {
        final TestingPhysicalSlot physicalSlot = TestingPhysicalSlot.builder().build();
        final AtomicBoolean externalReleaseInitiated = new AtomicBoolean(false);
        final SharedSlot sharedSlot =
                new SharedSlot(
                        new SlotRequestId(),
                        physicalSlot,
                        false,
                        () -> externalReleaseInitiated.set(true));

        sharedSlot.release(new Exception("test"));

        assertThat(externalReleaseInitiated).isFalse();
    }