[BESU-888] fix RocksDBException during sweeping (#929)

Co-authored-by: Ratan Rai Sur <ratan.r.sur@gmail.com>

Signed-off-by: Karim TAAM <karim.t2am@gmail.com>

plugins/rocksdb/src/main/java/org/hyperledger/besu/plugin/services/storage/rocksdb/segmented/RocksDBColumnarKeyValueStorage.java

@@ -234,6 +234,7 @@ public class RocksDBColumnarKeyValueStorage
     if (closed.compareAndSet(false, true)) {
       txOptions.close();
       options.close();
+      tryDeleteOptions.close();
       columnHandlesByName.values().forEach(ColumnFamilyHandle::close);
       db.close();
     }
@@ -298,7 +299,6 @@ public class RocksDBColumnarKeyValueStorage
     }
 
     private void close() {
-      tryDeleteOptions.close();
       innerTx.close();
       options.close();
     }

plugins/rocksdb/src/test/java/org/hyperledger/besu/plugin/services/storage/rocksdb/unsegmented/RocksDBColumnarKeyValueStorageTest.java

@@ -32,7 +32,6 @@ import java.util.Arrays;
 import java.util.Optional;
 import java.util.Set;
 
-import org.junit.Ignore;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.TemporaryFolder;
@@ -57,47 +56,55 @@ public class RocksDBColumnarKeyValueStorageTest extends AbstractKeyValueStorageT
         store.get(store.getSegmentIdentifierByName(TestSegment.FOO), bytesFromHexString("0001"));
 
     assertThat(result).isEmpty();
+
+    store.close();
   }
 
-  @Ignore
+  @Test
   public void canRemoveThroughSegmentIteration() throws Exception {
-    final SegmentedKeyValueStorage<ColumnFamilyHandle> store = createSegmentedStore();
-    final ColumnFamilyHandle fooSegment = store.getSegmentIdentifierByName(TestSegment.FOO);
-    final ColumnFamilyHandle barSegment = store.getSegmentIdentifierByName(TestSegment.BAR);
-
-    final Transaction<ColumnFamilyHandle> tx = store.startTransaction();
-    tx.put(fooSegment, bytesOf(1), bytesOf(1));
-    tx.put(fooSegment, bytesOf(2), bytesOf(2));
-    tx.put(fooSegment, bytesOf(3), bytesOf(3));
-    tx.put(barSegment, bytesOf(4), bytesOf(4));
-    tx.put(barSegment, bytesOf(5), bytesOf(5));
-    tx.put(barSegment, bytesOf(6), bytesOf(6));
-    tx.commit();
-
-    store
-        .streamKeys(fooSegment)
-        .forEach(
-            key -> {
-              if (!Arrays.equals(key, bytesOf(3))) store.tryDelete(fooSegment, key);
-            });
-    store
-        .streamKeys(barSegment)
-        .forEach(
-            key -> {
-              if (!Arrays.equals(key, bytesOf(4))) store.tryDelete(barSegment, key);
-            });
-
-    for (final ColumnFamilyHandle segment : Set.of(fooSegment, barSegment)) {
-      assertThat(store.streamKeys(segment).count()).isEqualTo(1);
+    // we're looping this in order to catch intermittent failures when rocksdb objects are not close
+    // properly
+    for (int i = 0; i < 50; i++) {
+      final SegmentedKeyValueStorage<ColumnFamilyHandle> store = createSegmentedStore();
+      final ColumnFamilyHandle fooSegment = store.getSegmentIdentifierByName(TestSegment.FOO);
+      final ColumnFamilyHandle barSegment = store.getSegmentIdentifierByName(TestSegment.BAR);
+
+      final Transaction<ColumnFamilyHandle> tx = store.startTransaction();
+      tx.put(fooSegment, bytesOf(1), bytesOf(1));
+      tx.put(fooSegment, bytesOf(2), bytesOf(2));
+      tx.put(fooSegment, bytesOf(3), bytesOf(3));
+      tx.put(barSegment, bytesOf(4), bytesOf(4));
+      tx.put(barSegment, bytesOf(5), bytesOf(5));
+      tx.put(barSegment, bytesOf(6), bytesOf(6));
+      tx.commit();
+
+      store
+          .streamKeys(fooSegment)
+          .forEach(
+              key -> {
+                if (!Arrays.equals(key, bytesOf(3))) store.tryDelete(fooSegment, key);
+              });
+      store
+          .streamKeys(barSegment)
+          .forEach(
+              key -> {
+                if (!Arrays.equals(key, bytesOf(4))) store.tryDelete(barSegment, key);
+              });
+
+      for (final ColumnFamilyHandle segment : Set.of(fooSegment, barSegment)) {
+        assertThat(store.streamKeys(segment).count()).isEqualTo(1);
+      }
+
+      assertThat(store.get(fooSegment, bytesOf(1))).isEmpty();
+      assertThat(store.get(fooSegment, bytesOf(2))).isEmpty();
+      assertThat(store.get(fooSegment, bytesOf(3))).contains(bytesOf(3));
+
+      assertThat(store.get(barSegment, bytesOf(4))).contains(bytesOf(4));
+      assertThat(store.get(barSegment, bytesOf(5))).isEmpty();
+      assertThat(store.get(barSegment, bytesOf(6))).isEmpty();
+
+      store.close();
     }
-
-    assertThat(store.get(fooSegment, bytesOf(1))).isEmpty();
-    assertThat(store.get(fooSegment, bytesOf(2))).isEmpty();
-    assertThat(store.get(fooSegment, bytesOf(3))).contains(bytesOf(3));
-
-    assertThat(store.get(barSegment, bytesOf(4))).contains(bytesOf(4));
-    assertThat(store.get(barSegment, bytesOf(5))).isEmpty();
-    assertThat(store.get(barSegment, bytesOf(6))).isEmpty();
   }
 
   @Test
@@ -129,6 +136,8 @@ public class RocksDBColumnarKeyValueStorageTest extends AbstractKeyValueStorageT
 
     assertThat(gotFromFoo).containsExactlyInAnyOrder(bytesOf(3));
     assertThat(gotFromBar).containsExactlyInAnyOrder(bytesOf(4), bytesOf(5));
+
+    store.close();
   }
 
   public enum TestSegment implements SegmentIdentifier {