[fix][client]Add setCompressMinMsgBodySize method to ProducerBuilder for compression configuration flexibility (#24164)

PIP: #23526

#### Motivation  
Currently, the `ProducerBuilder` API lacks the ability to directly configure the `compressMinMsgBodySize` property during producer initialization. Users are forced to modify the internal `ProducerConfigurationData` post-creation, which:  
1. **Violates encapsulation** by exposing internal configuration fields.  
2. **Introduces inconsistency** with other compression settings (e.g., `compressionType`) that are already configurable via the builder.  
3. **Creates usability hurdles**, requiring workarounds like:  
   ```java  
   Producer<byte[]> producer = client.newProducer().topic("test").create();  
   producer.conf.setCompressMinMsgBodySize(1024); // Insecure and error-prone  
   ```  
This change aligns the builder API with standard compression configuration patterns and improves developer ergonomics.  

#### Modifications  
1. **API Enhancement**:  
   - Added `compressMinMsgBodySize(int)` method to the `ProducerBuilder` interface.  
   - Propagate the configured value to `ProducerConfigurationData` during producer initialization.  
2. **Behavior Preservation**:  
   - Default value remains unchanged if the method is not invoked.  
   - Fully backward-compatible with existing code.  
3. **Testing & Documentation**:  
   - Added unit tests to validate compression threshold behavior via the new builder method.  
   - Updated Javadoc and configuration examples to reflect the new API.

Author: liangyepianzhou

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerConsumerInternalTest.java

@@ -245,15 +245,14 @@ public void testProducerCompressionMinMsgBodySize() throws PulsarClientException
                 .topic(topicName)
                 .producerName("producer")
                 .compressionType(CompressionType.LZ4)
+                .compressionMinMsgBodySize(1024)
                 .create();
         @Cleanup
         Consumer<byte[]> consumer = pulsarClient.newConsumer()
                 .topic(topicName)
                 .subscriptionName("sub")
                 .subscribe();
 
-        producer.conf.setCompressMinMsgBodySize(1024);
-        producer.conf.setCompressionType(CompressionType.LZ4);
         // disable batch
         producer.conf.setBatchingEnabled(false);
         producer.newMessage().value(msg1024).send();

pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ProducerBuilder.java

@@ -293,6 +293,19 @@ public interface ProducerBuilder<T> extends Cloneable {
      */
     ProducerBuilder<T> compressionType(CompressionType compressionType);
 
+    /**
+     * Sets the minimum uncompressed message body size required to enable compression.
+     * <p>
+     * When a message's body size exceeds this threshold (in bytes), compression will be applied
+     * using the configured {@link #compressionType(CompressionType)}. Messages smaller than this
+     * threshold will not be compressed.
+     * <p>
+     * Default: 4 KB
+     *
+     * @param compressionMinMsgBodySize the minimum uncompressed message body size required to enable compression
+     */
+    ProducerBuilder<T> compressionMinMsgBodySize(int compressionMinMsgBodySize);
+
     /**
      * Set a custom message routing policy by passing an implementation of MessageRouter.
      *

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBuilderImpl.java

@@ -174,6 +174,12 @@ public ProducerBuilder<T> compressionType(@NonNull CompressionType compressionTy
         return this;
     }
 
+    @Override
+    public ProducerBuilder<T> compressionMinMsgBodySize(int compressionMinMsgBodySize) {
+        conf.setCompressMinMsgBodySize(compressionMinMsgBodySize);
+        return this;
+    }
+
     @Override
     public ProducerBuilder<T> hashingScheme(@NonNull HashingScheme hashingScheme) {
         conf.setHashingScheme(hashingScheme);