String getLockName(String namespace, String name) {
        return "lock::" + namespace + "::" + kind() + "::" + name;
    }
@Test
    /*
     * Verifies that lock is released by call to `releaseLockAndTimer`. 
     * The call is made through a chain of futures ending with `eventually` after a failed execution via an unhandled exception in the `Callable`, 
     * followed by an unhandled exception occurring in the `onFailure` handler.
     */
    void testWithLockFailHandlerUnhandledExceptionReleasesLock(VertxTestContext context) {
        var resourceOperator = new DefaultWatchableStatusedResourceOperator<>(vertx, null, "TestResource");
        @SuppressWarnings({ "unchecked", "rawtypes" })
        var target = new DefaultOperator(vertx, "Test", resourceOperator, new MicrometerMetricsProvider(BackendRegistries.getDefaultNow()), null);
        Reconciliation reconciliation = new Reconciliation("test", "TestResource", "my-namespace", "my-resource");
        String lockName = target.getLockName(reconciliation);

        Promise<Void> handlersRegistered = Promise.promise();
        Promise<Void> failHandlerCalled = Promise.promise();

        @SuppressWarnings("unchecked")
        Future<String> result = target.withLockTest(reconciliation,
                // TEST SETUP: Do not throw the exception until all handlers registered
                () -> handlersRegistered.future().compose(nothing -> {
                    throw new UnsupportedOperationException(EXPECTED_MESSAGE);
                }));

        Checkpoint callableFailed = context.checkpoint();
        Checkpoint lockObtained  = context.checkpoint();

        result.onComplete(ar -> {
            assertThat(ar.failed(), is(true));
            Throwable e = ar.cause();
            context.verify(() -> {
                assertThat(e.getMessage(), is(EXPECTED_MESSAGE));
                callableFailed.flag();
            });
            try {
                throw new RuntimeException(e);
            } finally {
                // Enables the subsequent lock retrieval to verify it has been unlocked.
                failHandlerCalled.complete();
            }
        });

        failHandlerCalled.future()
            .compose(nothing ->
                vertx.sharedData().getLockWithTimeout(lockName, 10000L))
            .onComplete(context.succeeding(lock -> context.verify(() -> {
                assertThat(lock, instanceOf(Lock.class));
                lock.release();
                lockObtained.flag();
            })));

        handlersRegistered.complete();
    }