@ParameterizedTest
    @ArgumentsSource(SslTransportLayerArgumentsProvider.class)
    public void testServerTruststoreDynamicUpdate(Args args) throws Exception {
        SecurityProtocol securityProtocol = SecurityProtocol.SSL;
        args.sslServerConfigs.put(BrokerSecurityConfigs.SSL_CLIENT_AUTH_CONFIG, "required");
        TestSecurityConfig config = new TestSecurityConfig(args.sslServerConfigs);
        ListenerName listenerName = ListenerName.forSecurityProtocol(securityProtocol);
        ChannelBuilder serverChannelBuilder = ChannelBuilders.serverChannelBuilder(listenerName,
            false, securityProtocol, config, null, null, TIME, new LogContext(),
            defaultApiVersionsSupplier());
        server = new NioEchoServer(listenerName, securityProtocol, config,
                "localhost", serverChannelBuilder, null, TIME);
        server.start();
        InetSocketAddress addr = new InetSocketAddress("localhost", server.port());

        // Verify that client with matching keystore can authenticate, send and receive
        String oldNode = "0";
        Selector oldClientSelector = createSelector(args.sslClientConfigs);
        // take responsibility for closing oldClientSelector, so that we can keep it alive concurrent with the new one.
        this.selector = null;
        oldClientSelector.connect(oldNode, addr, BUFFER_SIZE, BUFFER_SIZE);
        NetworkTestUtils.checkClientConnection(oldClientSelector, oldNode, 100, 10);

        CertStores newClientCertStores = certBuilder(true, "client", args.useInlinePem).addHostName("localhost").build();
        args.sslClientConfigs = args.getTrustingConfig(newClientCertStores, args.serverCertStores);
        Map<String, Object> newTruststoreConfigs = newClientCertStores.trustStoreProps();
        assertInstanceOf(ListenerReconfigurable.class, serverChannelBuilder, "SslChannelBuilder not reconfigurable");
        ListenerReconfigurable reconfigurableBuilder = (ListenerReconfigurable) serverChannelBuilder;
        assertEquals(listenerName, reconfigurableBuilder.listenerName());
        reconfigurableBuilder.validateReconfiguration(newTruststoreConfigs);
        reconfigurableBuilder.reconfigure(newTruststoreConfigs);

        // Verify that new client with old truststore fails
        oldClientSelector.connect("1", addr, BUFFER_SIZE, BUFFER_SIZE);
        NetworkTestUtils.waitForChannelClose(oldClientSelector, "1", ChannelState.State.AUTHENTICATION_FAILED);

        // Verify that new client with new truststore can authenticate, send and receive
        Selector newClientSelector = createSelector(args.sslClientConfigs);
        newClientSelector.connect("2", addr, BUFFER_SIZE, BUFFER_SIZE);
        NetworkTestUtils.checkClientConnection(newClientSelector, "2", 100, 10);

        // Verify that old client continues to work
        NetworkTestUtils.checkClientConnection(oldClientSelector, oldNode, 100, 10);

        Map<String, Object>  invalidConfigs = new HashMap<>(newTruststoreConfigs);
        invalidConfigs.put(SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG, "INVALID_TYPE");
        verifyInvalidReconfigure(reconfigurableBuilder, invalidConfigs, "invalid truststore type");

        Map<String, Object>  missingStoreConfigs = new HashMap<>();
        missingStoreConfigs.put(SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG, "PKCS12");
        missingStoreConfigs.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, "some.truststore.path");
        missingStoreConfigs.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, new Password("some.truststore.password"));
        verifyInvalidReconfigure(reconfigurableBuilder, missingStoreConfigs, "truststore not found");

        // Verify that new connections continue to work with the server with previously configured keystore after failed reconfiguration
        newClientSelector.connect("3", addr, BUFFER_SIZE, BUFFER_SIZE);
        NetworkTestUtils.checkClientConnection(newClientSelector, "3", 100, 10);
        // manually stop the oldClientSelector because the test harness doesn't manage it.
        oldClientSelector.close();
    }
Tests reconfiguration of server truststore. Verifies that existing connections continue
to work with old truststore and new connections work with new truststore.
clients/src/test/java/org/apache/kafka/common/network/SslTransportLayerTest.java