@Test
    void verifyListener() throws Throwable {
        val registeredService = RegisteredServiceTestUtils.getRegisteredService();
        val publisherId = new PublisherIdentifier();
        val clientInfo = ClientInfoHolder.getClientInfo();
        casRegisteredServiceStreamPublisher.publish(registeredService,
            new CasRegisteredServiceSavedEvent(this, registeredService, clientInfo), publisherId);
        Thread.sleep(3000);
        assertFalse(registeredServiceDistributedCacheManager.getAll().isEmpty());

        casRegisteredServiceStreamPublisher.publish(registeredService,
            new CasRegisteredServiceDeletedEvent(this, registeredService, clientInfo), publisherId);

        Thread.sleep(2500);
        registeredServiceDistributedCacheManager.clear();
        assertTrue(registeredServiceDistributedCacheManager.getAll().isEmpty());
    }
This is {@link CasServicesStreamingKafkaConfigurationTests}.

@author Misagh Moayyed
@since 6.3.0
support/cas-server-support-service-registry-stream-kafka/src/test/java/org/apereo/cas/config/CasServicesStreamingKafkaConfigurationTests.java