@Test
    void verifyAction() throws Throwable {
        val registeredService = RegisteredServiceTestUtils.getRegisteredService();
        var obj = registeredServiceDistributedCacheManager.get(registeredService);
        assertNull(obj);
        assertFalse(registeredServiceDistributedCacheManager.contains(registeredService));

        val cache = DistributedCacheObject.<RegisteredService>builder()
            .value(registeredService)
            .publisherIdentifier(casRegisteredServiceStreamPublisherIdentifier)
            .build();

        registeredServiceDistributedCacheManager.set(registeredService, cache, true);
        Thread.sleep(2000);

        assertFalse(registeredServiceDistributedCacheManager.getAll().isEmpty());

        obj = registeredServiceDistributedCacheManager.get(registeredService);
        assertNotNull(obj);

        var c = registeredServiceDistributedCacheManager.findAll(obj1 -> obj1.getValue().equals(registeredService));
        assertFalse(c.isEmpty());

        registeredServiceDistributedCacheManager.remove(registeredService, cache, true);
        Thread.sleep(5000);
        c = registeredServiceDistributedCacheManager.findAll(obj1 -> obj1.getValue().equals(registeredService));
        assertTrue(c.isEmpty());
        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