public ConcurrentLongHashMap<CompletableFuture<Producer>> getProducers() {
        return producers;
    }
@Test(timeOut = 30000)
    public void testProducerFailureOnEncryptionRequiredOnBroker() throws Exception {
        // (a) Set encryption-required at broker level
        pulsar.getConfig().setEncryptionRequireOnProducer(true);
        resetChannel();
        setChannelConnected();

        // (b) Set encryption_required to false on policy
        Policies policies = mock(Policies.class);
        // Namespace policy doesn't require encryption
        policies.encryption_required = false;
        policies.topicDispatchRate = new HashMap<>();
        policies.clusterSubscribeRate = new HashMap<>();
        // add `clusterDispatchRate` otherwise there will be a NPE
        policies.clusterDispatchRate = new HashMap<>();
        // add `clusterDispatchRate` otherwise there will be a NPE
        // `org.apache.pulsar.broker.service.AbstractTopic.updateNamespaceSubscriptionDispatchRate`
        policies.subscriptionDispatchRate = new HashMap<>();
        // add `clusterDispatchRate` otherwise there will be a NPE
        // `org.apache.pulsar.broker.service.AbstractTopic.updateNamespaceReplicatorDispatchRate`
        policies.replicatorDispatchRate = new HashMap<>();
        pulsarTestContext.getPulsarResources().getNamespaceResources()
                .createPolicies(TopicName.get(encryptionRequiredTopicName).getNamespaceObject(),
                        policies);

        // test failure case: unencrypted producer cannot connect
        ByteBuf clientCommand =
                Commands.newProducer(encryptionRequiredTopicName, 2 /* producer id */, 2 /* request id */,
                        "unencrypted-producer", false, Collections.emptyMap(), false);
        channel.writeInbound(clientCommand);

        Object response = getResponse();
        assertEquals(response.getClass(), CommandError.class);
        CommandError errorResponse = (CommandError) response;
        assertEquals(errorResponse.getError(), ServerError.MetadataError);
        PersistentTopic topicRef = (PersistentTopic) brokerService.getTopicReference(encryptionRequiredTopicName).get();
        assertNotNull(topicRef);
        assertEquals(topicRef.getProducers().size(), 0);

        channel.finish();
    }