@Test
    public void testConnectionWithCustomKeyManager() throws Exception {
        TestProviderCreator testProviderCreator = new TestProviderCreator();

        int requestSize = 100 * 1024;
        final String node = "0";
        String request = TestUtils.randomString(requestSize);

        Map<String, Object> sslServerConfigs = TestSslUtils.createSslConfig(
                TestKeyManagerFactory.ALGORITHM,
                TestTrustManagerFactory.ALGORITHM,
                TestSslUtils.DEFAULT_TLS_PROTOCOL_FOR_TESTS
        );
        sslServerConfigs.put(SecurityConfig.SECURITY_PROVIDERS_CONFIG, testProviderCreator.getClass().getName());
        EchoServer server = new EchoServer(SecurityProtocol.SSL, sslServerConfigs);
        server.start();
        Time time = new MockTime();
        File trustStoreFile = new File(TestKeyManagerFactory.TestKeyManager.mockTrustStoreFile);
        Map<String, Object> sslClientConfigs = TestSslUtils.createSslConfig(true, true, ConnectionMode.CLIENT, trustStoreFile, "client");

        ChannelBuilder channelBuilder = new TestSslChannelBuilder(ConnectionMode.CLIENT);
        channelBuilder.configure(sslClientConfigs);
        Metrics metrics = new Metrics();
        Selector selector = new Selector(5000, metrics, time, "MetricGroup", channelBuilder, new LogContext());

        selector.connect(node, new InetSocketAddress("localhost", server.port), BUFFER_SIZE, BUFFER_SIZE);
        NetworkTestUtils.waitForChannelReady(selector, node);

        selector.send(createSend(node, request));

        waitForBytesBuffered(selector, node);

        TestUtils.waitForCondition(() -> cipherMetrics(metrics).size() == 1,
            "Waiting for cipher metrics to be created.");
        assertEquals(1, cipherMetrics(metrics).get(0).metricValue());
        assertNotNull(selector.channel(node).channelMetadataRegistry().cipherInformation());

        selector.close(node);
        super.verifySelectorEmpty(selector);

        assertEquals(1, cipherMetrics(metrics).size());
        assertEquals(0, cipherMetrics(metrics).get(0).metricValue());

        Security.removeProvider(testProviderCreator.getProvider().getName());
        selector.close();
        server.close();
        metrics.close();
    }
A set of tests for the selector. These use a test harness that runs a simple socket server that echos back responses.
clients/src/test/java/org/apache/kafka/common/network/SslSelectorTest.java