public <T> ProducerBuilder<T> createProducerBuilder(String topic, Schema<T> schema, String producerName) {
        ProducerBuilder<T> builder = client.newProducer(schema);
        if (defaultConfigurer != null) {
            defaultConfigurer.accept(builder);
        }
        builder.blockIfQueueFull(true)
                .enableBatching(true)
                .batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS)
                .hashingScheme(HashingScheme.Murmur3_32Hash) //
                .messageRoutingMode(MessageRoutingMode.CustomPartition)
                .messageRouter(FunctionResultRouter.of())
                // set send timeout to be infinity to prevent potential deadlock with consumer
                // that might happen when consumer is blocked due to unacked messages
                .sendTimeout(0, TimeUnit.SECONDS)
                .topic(topic);
        if (producerName != null) {
            builder.producerName(producerName);
        }
        if (producerConfig != null) {
            if (producerConfig.getCompressionType() != null) {
                builder.compressionType(producerConfig.getCompressionType());
            } else {
                // TODO: address this inconsistency.
                // PR https://github.com/apache/pulsar/pull/19470 removed the default compression type of LZ4
                // from the top level. This default is only used if producer config is provided.
                builder.compressionType(CompressionType.LZ4);
            }
            if (producerConfig.getMaxPendingMessages() != null && producerConfig.getMaxPendingMessages() != 0) {
                builder.maxPendingMessages(producerConfig.getMaxPendingMessages());
            }
            if (producerConfig.getMaxPendingMessagesAcrossPartitions() != null
                    && producerConfig.getMaxPendingMessagesAcrossPartitions() != 0) {
                builder.maxPendingMessagesAcrossPartitions(producerConfig.getMaxPendingMessagesAcrossPartitions());
            }
            if (producerConfig.getCryptoConfig() != null) {
                builder.cryptoKeyReader(crypto.keyReader);
                builder.cryptoFailureAction(crypto.failureAction);
                for (String encryptionKeyName : crypto.getEncryptionKeys()) {
                    builder.addEncryptionKey(encryptionKeyName);
                }
            }
            if (producerConfig.getBatchBuilder() != null) {
                if (producerConfig.getBatchBuilder().equals("KEY_BASED")) {
                    builder.batcherBuilder(BatcherBuilder.KEY_BASED);
                } else {
                    builder.batcherBuilder(BatcherBuilder.DEFAULT);
                }
            }
        }
        return builder;
    }
@Test
    public void testCreateProducerBuilder() {
        ProducerBuilderFactory builderFactory = new ProducerBuilderFactory(pulsarClient, null, null, null);
        builderFactory.createProducerBuilder("topic", Schema.STRING, "producerName");
        verifyCommon();
        verifyNoMoreInteractions(producerBuilder);
    }