[SPARK-45533][CORE] Use j.l.r.Cleaner instead of finalize for RocksDBIterator/LevelDBIterator

[zhaomin1423]

What changes were proposed in this pull request?

use java.lang.ref.Cleaner instead of finalize() for RocksDBIterator

Why are the changes needed?

The finalize() method has been marked as deprecated since Java 9 and will be removed in the future, java.lang.ref.Cleaner is the more recommended solution.

Does this PR introduce any user-facing change?

No

How was this patch tested?

Pass actions and new tests

Was this patch authored or co-authored using generative AI tooling?

No

[zhaomin1423]

cc @LuciferYang

[LuciferYang]

also cc @mridulm @beliefer @srowen @dongjoon-hyun

[LuciferYang]

Although the current pr only fixes part of RocksDB now, I suggest adding the LevelDB part in this PR once the RocksDB part is deemed feasible. I prefer to complete them in one.

[beliefer]

How do we finish the tests?

[zhaomin1423]

Although the current pr only fixes part of RocksDB now, I suggest adding the LevelDB part in this PR once the RocksDB part is deemed feasible. I prefer to complete them in one.

ok, will add LevelDB part later.

[zhaomin1423]

How do we finish the tests?

I haven’t thought of a good testing method yet, do you have any suggestions?

[LuciferYang]

@zhaomin1423

Firstly, we may need to add two helper methods:

1. Add a method in RocksDB that can get its corresponding WeakReference through RocksDBIterator, here I assume it's called iteratorReference
2. Add a method in RocksDBIterator to get RocksIterator, here I assume it's called internalIterator

Then, the test case may like the following:

@Test
public void testResourceCleaner() throws Exception {
  File dbPathForCleanerTest = File.createTempFile(
    "test_db_cleaner.", ".rdb");
  dbPathForCleanerTest.delete();

  RocksDB dbForCleanerTest = new RocksDB(dbPathForCleanerTest);
  for (int i = 0; i < 8192; i++) {
    dbForCleanerTest.write(createCustomType1(i));
  }

  RocksDBIterator<CustomType1> rocksDBIterator = (RocksDBIterator<CustomType1>) dbForCleanerTest.view(CustomType1.class).iterator();
  Optional<Reference<RocksDBIterator<?>>> referenceOpt = dbForCleanerTest.iteratorReference(rocksDBIterator);
  assertTrue(referenceOpt.isPresent());
  RocksIterator it = rocksDBIterator.internalIterator();
  assertTrue(it.isOwningHandle()); // it has not been closed yet, isOwningHandle should be true.
  rocksDBIterator = null; // Manually set rocksDBIterator to null, to be GC.
  Reference<RocksDBIterator<?>> ref = referenceOpt.get();
  // 100 times gc, the rocksDBIterator should be GCed.
  int count = 0;
  while (count < 100 && !ref.refersTo(null)) {
    System.gc();
    count++;
    Thread.sleep(100);
  }
  assertTrue(ref.refersTo(null)); // check rocksDBIterator should be GCed
  // Verify that the Cleaner will be executed after a period of time, and it.isOwningHandle() will become false.
  assertTimeout(java.time.Duration.ofSeconds(5), () -> assertFalse(it.isOwningHandle()));

  dbForCleanerTest.close();
}

Of course, we can also add a state in ResourceCleaner to indicate whether it has been executed, initially set to false, and turns true after execution. These are just some of my thoughts, there might be better ways to test.

[zhaomin1423]

@zhaomin1423

Firstly, we may need to add two helper methods:

1. Add a method in RocksDB that can get its corresponding WeakReference through RocksDBIterator, here I assume it's called iteratorReference
2. Add a method in RocksDBIterator to get RocksIterator, here I assume it's called internalIterator

Then, the test case may like the following:

@Test
public void testResourceCleaner() throws Exception {
  File dbPathForCleanerTest = File.createTempFile(
    "test_db_cleaner.", ".rdb");
  dbPathForCleanerTest.delete();

  RocksDB dbForCleanerTest = new RocksDB(dbPathForCleanerTest);
  for (int i = 0; i < 8192; i++) {
    dbForCleanerTest.write(createCustomType1(i));
  }

  RocksDBIterator<CustomType1> rocksDBIterator = (RocksDBIterator<CustomType1>) dbForCleanerTest.view(CustomType1.class).iterator();
  Optional<Reference<RocksDBIterator<?>>> referenceOpt = dbForCleanerTest.iteratorReference(rocksDBIterator);
  assertTrue(referenceOpt.isPresent());
  RocksIterator it = rocksDBIterator.internalIterator();
  assertTrue(it.isOwningHandle()); // it has not been closed yet, isOwningHandle should be true.
  rocksDBIterator = null; // Manually set rocksDBIterator to null, to be GC.
  Reference<RocksDBIterator<?>> ref = referenceOpt.get();
  // 100 times gc, the rocksDBIterator should be GCed.
  int count = 0;
  while (count < 100 && !ref.refersTo(null)) {
    System.gc();
    count++;
    Thread.sleep(100);
  }
  assertTrue(ref.refersTo(null)); // check rocksDBIterator should be GCed
  // Verify that the Cleaner will be executed after a period of time, and it.isOwningHandle() will become false.
  assertTimeout(java.time.Duration.ofSeconds(5), () -> assertFalse(it.isOwningHandle()));

  dbForCleanerTest.close();
}

Of course, we can also add a state in ResourceCleaner to indicate whether it has been executed, initially set to false, and turns true after execution. These are just some of my thoughts, there might be better ways to test.

Adding a state is simpler and works for different classes, but I don't know if adding variables for testing is a common approach.

[LuciferYang]

Let's add a test case first.

[LuciferYang]

Let's move dbForCleanerTest.close() to finally block and t would be better to add FileUtils.deleteQuietly(dbPathForCleanerTest); after close in finally block

[LuciferYang]

OK, it's fine as long as we can ensure there are no residues after testing.

[zhaomin1423]

sorry, if remove it, there is an exception.

org.rocksdb.RocksDBException: `/Users/zhaomin/Files/project/spark/common/kvstore/target/tmp/test_db_cleaner.6329921973463604622.rdb' exists but is not a directory

	at org.rocksdb.RocksDB.open(Native Method)
	at org.rocksdb.RocksDB.open(RocksDB.java:249)
	at org.apache.spark.util.kvstore.RocksDB.<init>(RocksDB.java:122)
	at org.apache.spark.util.kvstore.RocksDB.<init>(RocksDB.java:116)

[LuciferYang]

hmm.. this method can be moved to RocksDBSuite.java as a private method due to RocksDB already has getIteratorTracker(

[zhaomin1423]

done

[LuciferYang]

friendly ping @dongjoon-hyun @mridulm @beliefer Could you please take another look at this pr if you have time? Thanks ~

[LuciferYang]

Merged into master for Spark 4.0.
Thanks @zhaomin1423 ! Your work is greatly appreciated.
Thank you for your review! @mridulm @beliefer

[zhaomin1423]

Merged into master for Spark 4.0. Thanks @zhaomin1423 ! Your work is greatly appreciated. Thank you for your review! @mridulm @beliefer

Thank you very much for your help.