feat(java): configurable buffer size limit

java/fury-core/src/main/java/org/apache/fury/Fury.java

@@ -98,7 +98,6 @@ public final class Fury implements BaseFury {
   private static final byte isOutOfBandFlag = 1 << 3;
   private static final boolean isLittleEndian = ByteOrder.nativeOrder() == ByteOrder.LITTLE_ENDIAN;
   private static final byte BITMAP = isLittleEndian ? isLittleEndianFlag : 0;
-  private static final int BUFFER_SIZE_LIMIT = 128 * 1024;
   private static final short MAGIC_NUMBER = 0x62D4;
 
   private final Config config;
@@ -330,8 +329,8 @@ public MemoryBuffer getBuffer() {
 
   public void resetBuffer() {
     MemoryBuffer buf = buffer;
-    if (buf != null && buf.size() > BUFFER_SIZE_LIMIT) {
-      buffer = MemoryBuffer.newHeapBuffer(BUFFER_SIZE_LIMIT);
+    if (buf != null && buf.size() > config.bufferSizeLimitBytes()) {
+      buffer = MemoryBuffer.newHeapBuffer(config.bufferSizeLimitBytes());
     }
   }
 

java/fury-core/src/main/java/org/apache/fury/config/Config.java

@@ -61,6 +61,7 @@ public class Config implements Serializable {
   private transient int configHash;
   private final boolean deserializeNonexistentEnumValueAsNull;
   private final boolean serializeEnumByName;
+  private final int bufferSizeLimitBytes;
 
   public Config(FuryBuilder builder) {
     name = builder.name;
@@ -95,6 +96,7 @@ public Config(FuryBuilder builder) {
     scalaOptimizationEnabled = builder.scalaOptimizationEnabled;
     deserializeNonexistentEnumValueAsNull = builder.deserializeNonexistentEnumValueAsNull;
     serializeEnumByName = builder.serializeEnumByName;
+    bufferSizeLimitBytes = builder.bufferSizeLimitBytes;
   }
 
   /** Returns the name for Fury serialization. */
@@ -187,6 +189,10 @@ public LongEncoding longEncoding() {
     return longEncoding;
   }
 
+  public int bufferSizeLimitBytes() {
+    return bufferSizeLimitBytes;
+  }
+
   public boolean requireClassRegistration() {
     return requireClassRegistration;
   }
@@ -283,6 +289,7 @@ public boolean equals(Object o) {
         && compressString == config.compressString
         && compressInt == config.compressInt
         && compressLong == config.compressLong
+        && bufferSizeLimitBytes == config.bufferSizeLimitBytes
         && requireClassRegistration == config.requireClassRegistration
         && suppressClassRegistrationWarnings == config.suppressClassRegistrationWarnings
         && registerGuavaTypes == config.registerGuavaTypes
@@ -317,6 +324,7 @@ public int hashCode() {
         compressInt,
         compressLong,
         longEncoding,
+        bufferSizeLimitBytes,
         requireClassRegistration,
         suppressClassRegistrationWarnings,
         registerGuavaTypes,

java/fury-core/src/main/java/org/apache/fury/config/FuryBuilder.java

@@ -83,6 +83,7 @@ public final class FuryBuilder {
   boolean suppressClassRegistrationWarnings = true;
   boolean deserializeNonexistentEnumValueAsNull = false;
   boolean serializeEnumByName = false;
+  int bufferSizeLimitBytes = 128 * 1024;
   MetaCompressor metaCompressor = new DeflaterMetaCompressor();
 
   public FuryBuilder() {}
@@ -184,6 +185,17 @@ public FuryBuilder withStringCompressed(boolean stringCompressed) {
     return this;
   }
 
+  /**
+   * Sets the limit for Fury's internal buffer. If the buffer size exceeds this limit, it will be
+   * reset to this limit after every serialization and deserialization.
+   *
+   * <p>The default is 128k.
+   */
+  public FuryBuilder withBufferSizeLimitBytes(int bufferSizeLimitBytes) {
+    this.bufferSizeLimitBytes = bufferSizeLimitBytes;
+    return this;
+  }
+
   /**
    * Set classloader for fury to load classes, this classloader can't up updated. Fury will cache
    * the class meta data, if classloader can be updated, there may be class meta collision if

java/fury-core/src/test/java/org/apache/fury/FuryTest.java

@@ -615,4 +615,25 @@ public void testNullObjSerAndDe() {
     Object obj = fury.deserializeJavaObjectAndClass(bytes);
     assertNull(obj);
   }
+
+  @Test
+  public void testResetBufferToSizeLimit() {
+    final int minBufferBytes = 64;
+    final int limitInBytes = 1024;
+    Fury fury = Fury.builder().withBufferSizeLimitBytes(limitInBytes).build();
+
+    final byte[] smallPayload = new byte[0];
+    final byte[] serializedSmall = fury.serialize(smallPayload);
+    assertEquals(fury.getBuffer().size(), minBufferBytes);
+
+    fury.deserialize(serializedSmall);
+    assertEquals(fury.getBuffer().size(), minBufferBytes);
+
+    final byte[] largePayload = new byte[limitInBytes * 2];
+    final byte[] serializedLarge = fury.serialize(largePayload);
+    assertEquals(fury.getBuffer().size(), limitInBytes);
+
+    fury.deserialize(serializedLarge);
+    assertEquals(fury.getBuffer().size(), limitInBytes);
+  }
 }