fix(java): fix read primitives error on fill buffer bound (#2064)

## What does this PR do?
fix read primitives error on fill buffer bound
<!-- Describe the purpose of this PR. -->

## Related issues

Closes #2060 
## Does this PR introduce any user-facing change?

<!--
If any user-facing interface changes, please [open an
issue](https://github.com/apache/fury/issues/new/choose) describing the
need to do so and update the document if necessary.
-->

- [ ] Does this PR introduce any public API change?
- [ ] Does this PR introduce any binary protocol compatibility change?

## Benchmark

<!--
When the PR has an impact on performance (if you don't know whether the
PR will have an impact on performance, you can submit the PR first, and
if it will have impact on performance, the code reviewer will explain
it), be sure to attach a benchmark data here.
-->

Author: chaokunyang

java/fury-core/src/main/java/org/apache/fury/builder/ObjectCodecBuilder.java

@@ -625,14 +625,14 @@ private List<Expression> deserializeUnCompressedPrimitives(
     List<Expression> expressions = new ArrayList<>();
     int numPrimitiveFields = getNumPrimitiveFields(primitiveGroups);
     Literal totalSizeLiteral = Literal.ofInt(totalSize);
+    // After this check, following read can be totally unsafe without checks
+    expressions.add(new Invoke(buffer, "checkReadableBytes", totalSizeLiteral));
     Expression heapBuffer =
         new Invoke(buffer, "getHeapMemory", "heapBuffer", PRIMITIVE_BYTE_ARRAY_TYPE);
     Expression readerAddr =
         new Invoke(buffer, "getUnsafeReaderAddress", "readerAddr", PRIMITIVE_LONG_TYPE);
     expressions.add(heapBuffer);
     expressions.add(readerAddr);
-    // After this check, following read can be totally unsafe without checks
-    expressions.add(new Invoke(buffer, "checkReadableBytes", totalSizeLiteral));
     int acc = 0;
     for (List<Descriptor> group : primitiveGroups) {
       ListExpression groupExpressions = new ListExpression();
@@ -690,16 +690,17 @@ private List<Expression> deserializeCompressedPrimitives(
       Expression bean, Expression buffer, List<List<Descriptor>> primitiveGroups) {
     List<Expression> expressions = new ArrayList<>();
     int numPrimitiveFields = getNumPrimitiveFields(primitiveGroups);
-    Expression heapBuffer =
-        new Invoke(buffer, "getHeapMemory", "heapBuffer", PRIMITIVE_BYTE_ARRAY_TYPE);
-    expressions.add(heapBuffer);
     for (List<Descriptor> group : primitiveGroups) {
-      ListExpression groupExpressions = new ListExpression();
-      Expression readerAddr =
-          new Invoke(buffer, "getUnsafeReaderAddress", "readerAddr", PRIMITIVE_LONG_TYPE);
       // After this check, following read can be totally unsafe without checks.
+      // checkReadableBytes first, `fillBuffer` may create a new heap buffer.
       ReplaceStub checkReadableBytesStub = new ReplaceStub();
       expressions.add(checkReadableBytesStub);
+      Expression heapBuffer =
+          new Invoke(buffer, "getHeapMemory", "heapBuffer", PRIMITIVE_BYTE_ARRAY_TYPE);
+      expressions.add(heapBuffer);
+      ListExpression groupExpressions = new ListExpression();
+      Expression readerAddr =
+          new Invoke(buffer, "getUnsafeReaderAddress", "readerAddr", PRIMITIVE_LONG_TYPE);
       int acc = 0;
       boolean compressStarted = false;
       for (Descriptor descriptor : group) {

java/fury-core/src/main/java/org/apache/fury/io/FuryInputStream.java

@@ -76,15 +76,16 @@ public int fillBuffer(int minFillSize) {
 
   private static byte[] growBuffer(int minFillSize, MemoryBuffer buffer) {
     int newSize;
-    int targetSize = buffer.size() + minFillSize;
+    int size = buffer.size();
+    int targetSize = size + minFillSize;
     newSize =
         targetSize < MemoryBuffer.BUFFER_GROW_STEP_THRESHOLD
             ? targetSize << 2
             : (int) Math.min(targetSize * 1.5d, Integer.MAX_VALUE - 8);
     byte[] newBuffer = new byte[newSize];
     byte[] heapMemory = buffer.getHeapMemory();
-    System.arraycopy(heapMemory, 0, newBuffer, 0, buffer.size());
-    buffer.initHeapBuffer(newBuffer, 0, buffer.size());
+    System.arraycopy(heapMemory, 0, newBuffer, 0, size);
+    buffer.initHeapBuffer(newBuffer, 0, size);
     heapMemory = newBuffer;
     return heapMemory;
   }

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

@@ -391,4 +391,26 @@ public void testReadNullChunkMapOnFillBound() {
     FuryInputStream input = new FuryInputStream(inputStream);
     assertEquals(fury.deserialize(input), m);
   }
+
+  public static class SimpleType {
+    public double dVal;
+
+    public SimpleType() {
+      dVal = 0.5;
+    }
+  }
+
+  // For issue https://github.com/apache/fury/issues/2060
+  @Test
+  public void testReadPrimitivesOnBufferFillBound() {
+    Fury fury = builder().build();
+    fury.register(SimpleType.class);
+    SimpleType v = new SimpleType();
+    ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+    fury.serialize(outputStream, v);
+    InputStream inputStream = new ByteArrayInputStream(outputStream.toByteArray());
+    FuryInputStream input = new FuryInputStream(inputStream, 11);
+    SimpleType newValue = (SimpleType) fury.deserialize(input);
+    Assert.assertEquals(v.dVal, newValue.dVal, 0.001);
+  }
 }