@VisibleForTesting
    protected void tombstone() {
        var now = System.currentTimeMillis();
        if (now - lastTombstonedAt < tombstoneDelayInMillis) {
            return;
        }
        var lastSuccessfulTombstonedAt = lastTombstonedAt;
        lastTombstonedAt = now; // dedup first
        bundleLoadDataStore.removeAsync(brokerId)
                .whenComplete((__, e) -> {
                            if (e != null) {
                                log.error("Failed to clean broker load data.", e);
                                lastTombstonedAt = lastSuccessfulTombstonedAt;
                            } else {
                                boolean debug = ExtensibleLoadManagerImpl.debug(pulsar.getConfiguration(), log);
                                if (debug) {
                                    log.info("Cleaned broker load data.");
                                }
                            }
                        }
                );
    }
@Test
    public void testTombstone() throws IllegalAccessException {

        var target = spy(new TopBundleLoadDataReporter(pulsar, broker, store));

        target.handleEvent(bundle,
                new ServiceUnitStateData(ServiceUnitState.Assigning, broker, VERSION_ID_INIT), null);
        verify(store, times(0)).removeAsync(eq(broker));
        verify(target, times(0)).tombstone();

        target.handleEvent(bundle,
                new ServiceUnitStateData(ServiceUnitState.Deleted, broker, VERSION_ID_INIT), null);
        verify(store, times(0)).removeAsync(eq(broker));
        verify(target, times(0)).tombstone();


        target.handleEvent(bundle,
                new ServiceUnitStateData(ServiceUnitState.Init, broker, VERSION_ID_INIT), null);
        verify(store, times(0)).removeAsync(eq(broker));
        verify(target, times(0)).tombstone();

        target.handleEvent(bundle,
                new ServiceUnitStateData(ServiceUnitState.Free, broker, VERSION_ID_INIT), null);
        verify(store, times(0)).removeAsync(eq(broker));
        verify(target, times(0)).tombstone();

        target.handleEvent(bundle,
                new ServiceUnitStateData(ServiceUnitState.Releasing, "broker-2", broker, VERSION_ID_INIT),
                new RuntimeException());
        verify(store, times(0)).removeAsync(eq(broker));
        verify(target, times(0)).tombstone();

        target.handleEvent(bundle,
                new ServiceUnitStateData(ServiceUnitState.Releasing, "broker-2", broker, VERSION_ID_INIT), null);
        Awaitility.waitAtMost(3, TimeUnit.SECONDS).untilAsserted(() -> {
            verify(target, times(1)).tombstone();
            verify(store, times(1)).removeAsync(eq(broker));
        });

        target.handleEvent(bundle,
                new ServiceUnitStateData(ServiceUnitState.Releasing, "broker-2", broker, VERSION_ID_INIT), null);
        Awaitility.waitAtMost(3, TimeUnit.SECONDS).untilAsserted(() -> {
            verify(target, times(2)).tombstone();
            verify(store, times(1)).removeAsync(eq(broker));
        });

        FieldUtils.writeDeclaredField(target, "tombstoneDelayInMillis", 0, true);
        target.handleEvent(bundle,
                new ServiceUnitStateData(ServiceUnitState.Splitting, "broker-2", broker, VERSION_ID_INIT), null);
        Awaitility.waitAtMost(3, TimeUnit.SECONDS).untilAsserted(() -> {
            verify(target, times(3)).tombstone();
            verify(store, times(2)).removeAsync(eq(broker));
        });

        target.handleEvent(bundle,
                new ServiceUnitStateData(ServiceUnitState.Owned, broker, VERSION_ID_INIT), null);
        Awaitility.waitAtMost(3, TimeUnit.SECONDS).untilAsserted(() -> {
            verify(target, times(4)).tombstone();
            verify(store, times(3)).removeAsync(eq(broker));
        });
    }