@Override
    public void onMetadataUpdate(
        MetadataDelta delta,
        MetadataImage newImage,
        LoaderManifest manifest
    ) {
        boolean checkBrokerRegistration = false;
        if (delta.featuresDelta() != null) {
            if (delta.metadataVersionChanged().isPresent()) {
                if (log.isTraceEnabled()) {
                    log.trace("Metadata version change is present: {}",
                        delta.metadataVersionChanged());
                }
                checkBrokerRegistration = true;
            }
        }
        if (delta.clusterDelta() != null) {
            if (delta.clusterDelta().changedBrokers().get(id) != null) {
                if (log.isTraceEnabled()) {
                    log.trace("Broker change is present: {}",
                        delta.clusterDelta().changedBrokers().get(id));
                }
                checkBrokerRegistration = true;
            }
        }
        if (checkBrokerRegistration) {
            if (brokerRegistrationNeedsRefresh(newImage.features().metadataVersion(),
                    delta.clusterDelta().broker(id))) {
                refreshRegistrationCallback.run();
            }
        }
    }
@Test
    public void testBrokerUpdateWithoutNewMvDoesNothing() {
        BrokerRegistrationTrackerTestContext ctx  = new BrokerRegistrationTrackerTestContext();
        MetadataDelta delta = ctx.newDelta();
        delta.replay(new RegisterBrokerRecord().
            setBrokerId(1).
            setIncarnationId(INCARNATION_ID).
            setLogDirs(Arrays.asList(A, B, C)));
        ctx.onMetadataUpdate(delta);
        assertEquals(0, ctx.numCalls.get());
    }