fix(java): ClassLoaderFuryPooled#setFactoryCallback cannot effect old…

… Fury (#1946)



## What does this PR do?

ClassLoaderFuryPooled#setFactoryCallback cannot effect old fury. 
so if
`org.apache.fury.pool.FuryPooledObjectFactory#classLoaderFuryPooledCache`
expired, new classLoaderFuryPooled can not effected by custom
factoryCallback

<!-- Describe the purpose of this PR. -->

## Related issues

<!--
Is there any related issue? Please attach here.

- #xxxx0
- #xxxx1
- #xxxx2
-->

## 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.
-->

Co-authored-by: shuchang.li <[email protected]>

java/fury-core/src/main/java/org/apache/fury/pool/ClassLoaderFuryPooled.java

@@ -123,6 +123,7 @@ private void addFury() {
   }
 
   void setFactoryCallback(Consumer<Fury> factoryCallback) {
-    this.factoryCallback = factoryCallback;
+    this.factoryCallback = this.factoryCallback.andThen(factoryCallback);
+    allFury.keySet().forEach(factoryCallback);
   }
 }

java/fury-core/src/test/java/org/apache/fury/pool/ClassLoaderFuryPooledTest.java

@@ -83,6 +83,23 @@ public void testGetFuryWithIncreaseCapacity() {
     }
   }
 
+  @Test
+  public void testFuryAfterSetFactoryCallback() {
+    int minPoolSize = 4;
+    ClassLoaderFuryPooled pooled = getPooled(minPoolSize, 6);
+
+    try {
+      pooled.setFactoryCallback(
+          fury -> {
+            throw new RuntimeException();
+          });
+      pooled.getFury();
+      Assert.fail();
+    } catch (RuntimeException e) {
+      // Success
+    }
+  }
+
   @Test
   public void testGetFuryAwait() throws InterruptedException {
     int minPoolSize = 3;