public void closeOperation(OperationHandle operationHandle) {
        writeLock(
                () -> {
                    Operation opToRemove = submittedOperations.remove(operationHandle);
                    if (opToRemove != null) {
                        opToRemove.close();
                    }
                });
    }
@Test
    void testCloseOperation() {
        CountDownLatch endRunningLatch = new CountDownLatch(1);
        OperationHandle operationHandle =
                operationManager.submitOperation(
                        () -> {
                            endRunningLatch.await();
                            return defaultResultSet;
                        });

        EXECUTOR_EXTENSION
                .getExecutor()
                .submit(() -> operationManager.closeOperation(operationHandle));

        assertThatThrownBy(
                        () -> {
                            operationManager.awaitOperationTermination(operationHandle);
                            operationManager.getOperation(operationHandle);
                        })
                .satisfies(
                        FlinkAssertions.anyCauseMatches(
                                SqlGatewayException.class,
                                String.format(
                                        "Can not find the submitted operation in the OperationManager with the %s.",
                                        operationHandle)));
    }