@After
public void cleanupRocksDB() {
	if (keyedStateBackend != null) {
		IOUtils.closeQuietly(keyedStateBackend);
		keyedStateBackend.dispose();
	}
	IOUtils.closeQuietly(defaultCFHandle);
	IOUtils.closeQuietly(db);
	IOUtils.closeQuietly(columnOptions);
	IOUtils.closeQuietly(dbOptions);

	if (allCreatedCloseables != null) {
		for (RocksObject rocksCloseable : allCreatedCloseables) {
			verify(rocksCloseable, times(1)).close();
		}
		allCreatedCloseables = null;
	}
	try {
		org.apache.flink.util.FileUtils.deleteDirectory(instanceBasePath);
	} catch (Exception ex) {
		// ignored
	}
}
 

private void verifyRocksObjectsReleased() {
	//Ensure every RocksObject was closed exactly once
	for (RocksObject rocksCloseable : allCreatedCloseables) {
		verify(rocksCloseable, times(1)).close();
	}

	assertNotNull(null, keyedStateBackend.db);
	RocksDB spyDB = keyedStateBackend.db;

	if (!enableIncrementalCheckpointing) {
		verify(spyDB, times(1)).getSnapshot();
		verify(spyDB, times(1)).releaseSnapshot(any(Snapshot.class));
	}

	keyedStateBackend.dispose();
	verify(spyDB, times(1)).close();
	assertEquals(true, keyedStateBackend.isDisposed());
}
 

@After
public void cleanupRocksDB() {
	if (keyedStateBackend != null) {
		IOUtils.closeQuietly(keyedStateBackend);
		keyedStateBackend.dispose();
	}
	IOUtils.closeQuietly(defaultCFHandle);
	IOUtils.closeQuietly(db);
	IOUtils.closeQuietly(columnOptions);
	IOUtils.closeQuietly(dbOptions);

	if (allCreatedCloseables != null) {
		for (RocksObject rocksCloseable : allCreatedCloseables) {
			verify(rocksCloseable, times(1)).close();
		}
		allCreatedCloseables = null;
	}
	try {
		org.apache.flink.util.FileUtils.deleteDirectory(instanceBasePath);
	} catch (Exception ex) {
		// ignored
	}
}
 

private void verifyRocksObjectsReleased() {
	//Ensure every RocksObject was closed exactly once
	for (RocksObject rocksCloseable : allCreatedCloseables) {
		verify(rocksCloseable, times(1)).close();
	}

	assertNotNull(null, keyedStateBackend.db);
	RocksDB spyDB = keyedStateBackend.db;

	if (!enableIncrementalCheckpointing) {
		verify(spyDB, times(1)).getSnapshot();
		verify(spyDB, times(1)).releaseSnapshot(any(Snapshot.class));
	}

	keyedStateBackend.dispose();
	verify(spyDB, times(1)).close();
	assertEquals(true, keyedStateBackend.isDisposed());
}
 

@After
public void cleanupRocksDB() {
	if (keyedStateBackend != null) {
		IOUtils.closeQuietly(keyedStateBackend);
		keyedStateBackend.dispose();
	}
	IOUtils.closeQuietly(defaultCFHandle);
	IOUtils.closeQuietly(db);
	IOUtils.closeQuietly(optionsContainer);

	if (allCreatedCloseables != null) {
		for (RocksObject rocksCloseable : allCreatedCloseables) {
			verify(rocksCloseable, times(1)).close();
		}
		allCreatedCloseables = null;
	}
}
 

private void verifyRocksObjectsReleased() {
	//Ensure every RocksObject was closed exactly once
	for (RocksObject rocksCloseable : allCreatedCloseables) {
		verify(rocksCloseable, times(1)).close();
	}

	assertNotNull(null, keyedStateBackend.db);
	RocksDB spyDB = keyedStateBackend.db;

	if (!enableIncrementalCheckpointing) {
		verify(spyDB, times(1)).getSnapshot();
		verify(spyDB, times(1)).releaseSnapshot(any(Snapshot.class));
	}

	keyedStateBackend.dispose();
	verify(spyDB, times(1)).close();
	assertEquals(true, keyedStateBackend.isDisposed());
}
 

private static <T extends RocksObject> T checkInvalid(final T opts) {
    if (!opts.isOwningHandle()) {
        throw new IllegalStateException(
            "the instance of options [" + opts
                    + "] has been released, calling any of its functions will lead to undefined behavior.");
    }
    return opts;
}